package cql

import (
	"bytes"
	"context"
	"encoding/binary"
	"fmt"
	"regexp"
	"strings"

	"go.uber.org/zap"
)

// ============================================================================
// Protocol Version Management
// ============================================================================

// ProtocolVersionContext manages protocol version information within a context.
// This abstracts away the repetitive protocol version extraction logic.
type ProtocolVersionContext struct {
	version byte
}

// NewProtocolVersionContext creates a new protocol version context.
func NewProtocolVersionContext(version byte) *ProtocolVersionContext {
	return &ProtocolVersionContext{version: version}
}

// GetVersion returns the protocol version.
func (p *ProtocolVersionContext) GetVersion() byte {
	return p.version
}

// GetResponseVersion returns the response version for the current protocol version.
// Response versions have the high bit set (0x80).
func (p *ProtocolVersionContext) GetResponseVersion() byte {
	return ResponseVersion(p.version)
}

// SupportsSegmentFormat returns true if the protocol version uses segment format (v5+).
func (p *ProtocolVersionContext) SupportsSegmentFormat() bool {
	return p.version >= ProtoVersion5
}

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

// ExtractProtocolVersion extracts protocol version from a frame or connection context.
// It prioritizes connection context over frame version.
func ExtractProtocolVersion(ctx context.Context, conn *Conn, frameVersion byte) byte {
	if conn != nil {
		return conn.GetProtocolVersion()
	}
	return normalizeProtoVersionByte(frameVersion)
}

// ============================================================================
// Stream ID Validation
// ============================================================================

// StreamIDValidator validates stream IDs based on protocol version.
type StreamIDValidator struct {
	maxStreamID int16
	version     byte
}

// NewStreamIDValidator creates a new stream ID validator.
func NewStreamIDValidator(version byte) *StreamIDValidator {
	var maxStreamID int16
	if version <= ProtoVersion2 {
		maxStreamID = MaxStreamIDv1v2
	} else {
		maxStreamID = MaxStreamIDv3p
	}
	return &StreamIDValidator{
		maxStreamID: maxStreamID,
		version:     version,
	}
}

// Validate checks if the stream ID is valid for the protocol version.
// Returns an error if the stream ID is out of range.
func (v *StreamIDValidator) Validate(stream int16) error {
	if stream > v.maxStreamID || stream < 0 {
		return fmt.Errorf("invalid stream ID %d for protocol version %d (max: %d)", stream, v.version, v.maxStreamID)
	}
	return nil
}

// ============================================================================
// Authentication Check
// ============================================================================

// AuthChecker checks authentication requirements for CQL operations.
type AuthChecker struct {
	authenticator Authenticator
}

// NewAuthChecker creates a new authentication checker.
func NewAuthChecker(authenticator Authenticator) *AuthChecker {
	return &AuthChecker{authenticator: authenticator}
}

// IsAuthenticationRequired checks if authentication is required for the given opcode.
// STARTUP and OPTIONS messages don't require authentication (they are allowed without auth).
// Other messages require authentication if no user is logged in and the authenticator requires auth.
func (ac *AuthChecker) IsAuthenticationRequired(opcode byte, username string) bool {
	// STARTUP/OPTIONS and AUTH_RESPONSE are always allowed without prior authentication.
	// AUTH_RESPONSE must pass through so SASL login can complete.
	if opcode == OpcodeStartup || opcode == OpcodeOptions || opcode == OpcodeAuthResponse {
		return false
	}
	// For other messages, check if we need authentication
	if username == "" && ac.authenticator != nil && ac.authenticator.RequiresAuth() {
		return true
	}
	return false
}

// ShouldAuthenticateStartup checks if STARTUP should trigger AUTHENTICATE response.
// This is separate from IsAuthenticationRequired because STARTUP is always allowed
// but may require authentication based on the authenticator configuration.
func (ac *AuthChecker) ShouldAuthenticateStartup() bool {
	return ac.authenticator != nil && ac.authenticator.RequiresAuth()
}

// ============================================================================
// Error Frame Creation
// ============================================================================

// ErrorFrameBuilder creates error response frames.
type ErrorFrameBuilder struct {
	version byte
}

var protocolBackendErrorCodeRE = regexp.MustCompile(`\[[A-Za-z_]+:\d+\]\s*`)
var protocolBackendErrorPrefixRE = regexp.MustCompile(`(?i)(^|[:;]\s*)(?:error|errno)\s+\d+(?:\s*\([0-9A-Z]+\))?:\s*`)
var protocolBackendKeywordRE = regexp.MustCompile(`(?i)(^|[^[:alnum:]_'"\x60])(tidb|mysql|tikv)([^[:alnum:]_'"\x60]|$)`)
var protocolBackendWhitespaceRE = regexp.MustCompile(`\s+`)

// NewErrorFrameBuilder creates a new error frame builder.
func NewErrorFrameBuilder(version byte) *ErrorFrameBuilder {
	return &ErrorFrameBuilder{version: version}
}

// Build creates an error frame with the given error code and message.
func (b *ErrorFrameBuilder) Build(stream int16, errorCode int, message string) *Frame {
	writer := NewFrameBodyWriter()
	_ = writer.WriteInt(int32(errorCode))
	_ = writer.WriteString(sanitizeProtocolErrorMessage(message))
	writeErrorDetails(writer, errorCode)

	// Use response version based on current protocol version
	responseVersion := ResponseVersion(b.version)

	return &Frame{
		Version: responseVersion,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeError,
		Length:  int32(len(writer.Bytes())),
		Body:    writer.Bytes(),
	}
}

// BuildUnprepared creates an unprepared error frame with the missing statement ID.
func (b *ErrorFrameBuilder) BuildUnprepared(stream int16, message string, stmtID string) *Frame {
	writer := NewFrameBodyWriter()
	_ = writer.WriteInt(int32(ErrorCodeUnprepared))
	_ = writer.WriteString(sanitizeProtocolErrorMessage(message))
	_ = writer.WriteShortBytes([]byte(stmtID))

	responseVersion := ResponseVersion(b.version)
	return &Frame{
		Version: responseVersion,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeError,
		Length:  int32(len(writer.Bytes())),
		Body:    writer.Bytes(),
	}
}

func writeErrorDetails(writer *FrameBodyWriter, errorCode int) {
	switch errorCode {
	case ErrorCodeUnavailable:
		_ = writer.WriteConsistency(ConsistencyOne)
		_ = writer.WriteInt(0)
		_ = writer.WriteInt(0)
	case ErrorCodeWriteTimeout:
		_ = writer.WriteConsistency(ConsistencyOne)
		_ = writer.WriteInt(0)
		_ = writer.WriteInt(0)
		_ = writer.WriteString("SIMPLE")
	case ErrorCodeReadTimeout:
		_ = writer.WriteConsistency(ConsistencyOne)
		_ = writer.WriteInt(0)
		_ = writer.WriteInt(0)
		_ = writer.WriteByte(0)
	case ErrorCodeAlreadyExists:
		_ = writer.WriteString("")
		_ = writer.WriteString("")
	case ErrorCodeUnprepared:
		_ = writer.WriteShortBytes([]byte{})
	}
}

// CreateErrorFrame is a convenience function to create an error frame.
// It extracts the protocol version from the context or connection.
func CreateErrorFrame(ctx context.Context, conn *Conn, stream int16, errorCode int, message string) *Frame {
	version := ExtractProtocolVersion(ctx, conn, 0)
	builder := NewErrorFrameBuilder(version)
	return builder.Build(stream, errorCode, message)
}

func sanitizeProtocolErrorMessage(message string) string {
	sanitized := strings.TrimSpace(message)
	if sanitized == "" {
		return ""
	}

	sanitized = protocolBackendErrorPrefixRE.ReplaceAllString(sanitized, "${1}")
	sanitized = protocolBackendErrorCodeRE.ReplaceAllString(sanitized, "")
	sanitized = protocolBackendKeywordRE.ReplaceAllString(sanitized, "${1}backend${3}")
	sanitized = protocolBackendWhitespaceRE.ReplaceAllString(sanitized, " ")
	sanitized = strings.TrimSpace(sanitized)
	sanitized = strings.TrimLeft(sanitized, ":;- ")
	return strings.TrimSpace(sanitized)
}

// ============================================================================
// Frame Body Parsing Utilities
// ============================================================================

// FrameBodyParser provides utilities for parsing CQL frame bodies.
type FrameBodyParser struct {
	data []byte
	pos  int
}

// NewFrameBodyParser creates a new frame body parser.
func NewFrameBodyParser(data []byte) *FrameBodyParser {
	return &FrameBodyParser{data: data, pos: 0}
}

// Pos returns the current position in the data.
func (p *FrameBodyParser) Pos() int {
	return p.pos
}

// HasMore returns true if there is more data to parse.
func (p *FrameBodyParser) HasMore() bool {
	return p.pos < len(p.data)
}

// ReadValue reads a length-prefixed value (int32 length) and distinguishes NULL vs UNSET.
func (p *FrameBodyParser) ReadValue() ([]byte, bool, bool, error) {
	if len(p.data) < p.pos+4 {
		return nil, false, false, fmt.Errorf("not enough bytes to read bytes header")
	}

	length := int32(binary.BigEndian.Uint32(p.data[p.pos : p.pos+4]))
	p.pos += 4

	switch length {
	case NullValue:
		return nil, true, false, nil
	case UnsetValue:
		return nil, false, true, nil
	}
	if length < 0 {
		return nil, false, false, fmt.Errorf("invalid bytes length %d", length)
	}

	if len(p.data) < p.pos+int(length) {
		return nil, false, false, fmt.Errorf("not enough bytes to read bytes value")
	}

	result := p.data[p.pos : p.pos+int(length)]
	p.pos += int(length)
	return result, false, false, nil
}

// ReadBytes reads a length-prefixed byte slice (int32 length).
func (p *FrameBodyParser) ReadBytes() ([]byte, error) {
	value, _, _, err := p.ReadValue()
	return value, err
}

// ReadBytesN reads an exact number of bytes.
func (p *FrameBodyParser) ReadBytesN(n int) ([]byte, error) {
	if n < 0 {
		return nil, fmt.Errorf("invalid bytes length %d", n)
	}
	if len(p.data) < p.pos+n {
		return nil, fmt.Errorf("not enough bytes to read %d bytes", n)
	}
	result := p.data[p.pos : p.pos+n]
	p.pos += n
	return result, nil
}

// ReadShortBytes reads a length-prefixed byte slice (uint16 length).
func (p *FrameBodyParser) ReadShortBytes() ([]byte, error) {
	if len(p.data) < p.pos+2 {
		return nil, fmt.Errorf("not enough bytes to read short bytes header")
	}

	length := int(binary.BigEndian.Uint16(p.data[p.pos : p.pos+2]))
	p.pos += 2

	if len(p.data) < p.pos+length {
		return nil, fmt.Errorf("not enough bytes to read short bytes value")
	}

	result := p.data[p.pos : p.pos+length]
	p.pos += length
	return result, nil
}

// ReadString reads a length-prefixed string (uint16 length).
func (p *FrameBodyParser) ReadString() (string, error) {
	bytes, err := p.ReadShortBytes()
	if err != nil {
		return "", err
	}
	return string(bytes), nil
}

// ReadLongString reads a length-prefixed string (int32 length).
func (p *FrameBodyParser) ReadLongString() (string, error) {
	if len(p.data) < p.pos+4 {
		return "", fmt.Errorf("not enough bytes to read long string header")
	}
	str, n, err := readLongString(p.data[p.pos:])
	if err != nil {
		return "", err
	}
	p.pos += n
	return str, nil
}

// ReadInt reads an int32 value.
func (p *FrameBodyParser) ReadInt() (int32, error) {
	if len(p.data) < p.pos+4 {
		return 0, fmt.Errorf("not enough bytes to read int")
	}
	val := int32(binary.BigEndian.Uint32(p.data[p.pos : p.pos+4]))
	p.pos += 4
	return val, nil
}

// ReadShort reads a uint16 value.
func (p *FrameBodyParser) ReadShort() (uint16, error) {
	if len(p.data) < p.pos+2 {
		return 0, fmt.Errorf("not enough bytes to read short")
	}
	val := binary.BigEndian.Uint16(p.data[p.pos : p.pos+2])
	p.pos += 2
	return val, nil
}

// ReadByte reads a single byte.
func (p *FrameBodyParser) ReadByte() (byte, error) {
	if len(p.data) < p.pos+1 {
		return 0, fmt.Errorf("not enough bytes to read byte")
	}
	val := p.data[p.pos]
	p.pos++
	return val, nil
}

// ReadConsistency reads a consistency level (uint16).
func (p *FrameBodyParser) ReadConsistency() (uint16, error) {
	return p.ReadShort()
}

// ReadFlags reads protocol version-specific flags.
// For v5+, flags is uint32; for v3-v4, flags is byte.
func (p *FrameBodyParser) ReadFlags(protoVer byte) (uint32, error) {
	if protoVer >= ProtoVersion5 {
		if len(p.data) < p.pos+4 {
			return 0, fmt.Errorf("not enough bytes to read flags")
		}
		flags := binary.BigEndian.Uint32(p.data[p.pos : p.pos+4])
		p.pos += 4
		return flags, nil
	}

	if len(p.data) < p.pos+1 {
		return 0, fmt.Errorf("not enough bytes to read flags")
	}
	flags := uint32(p.data[p.pos])
	p.pos++
	return flags, nil
}

// ReadValueList reads query values (positional or named) based on flags.
func (p *FrameBodyParser) ReadValueList(flags uint32) ([]QueryValue, bool, error) {
	if flags&QueryFlagWithNameValues != 0 && flags&QueryFlagValues == 0 {
		return nil, false, fmt.Errorf("named values require value list flag")
	}
	if flags&QueryFlagValues == 0 {
		return nil, false, nil
	}

	valueCount, err := p.ReadShort()
	if err != nil {
		return nil, false, err
	}

	named := flags&QueryFlagWithNameValues != 0
	values := make([]QueryValue, valueCount)
	for i := uint16(0); i < valueCount; i++ {
		if named {
			name, err := p.ReadString()
			if err != nil {
				return nil, false, err
			}
			values[i].Name = name
		}
		value, _, isUnset, err := p.ReadValue()
		if err != nil {
			return nil, false, err
		}
		values[i].Value = value
		values[i].Unset = isUnset
	}

	return values, named, nil
}

// Skip skips n bytes.
func (p *FrameBodyParser) Skip(n int) error {
	if len(p.data) < p.pos+n {
		return fmt.Errorf("not enough bytes to skip %d bytes", n)
	}
	p.pos += n
	return nil
}

// ============================================================================
// Credential Extraction
// ============================================================================

// CredentialExtractor extracts credentials from a connection context.
type CredentialExtractor struct{}

// NewCredentialExtractor creates a new credential extractor.
func NewCredentialExtractor() *CredentialExtractor {
	return &CredentialExtractor{}
}

// Extract extracts username and password from the connection in the context.
func (e *CredentialExtractor) Extract(ctx context.Context) (username string, password string) {
	conn := connFromContext(ctx)
	if conn != nil {
		username, password = conn.GetCredentials()
	}
	return username, password
}

// ============================================================================
// Frame Body Writer
// ============================================================================

// FrameBodyWriter provides utilities for building CQL frame bodies.
type FrameBodyWriter struct {
	buf bytes.Buffer
}

// NewFrameBodyWriter creates a new frame body writer.
func NewFrameBodyWriter() *FrameBodyWriter {
	return &FrameBodyWriter{}
}

// WriteByte writes a single byte.
func (w *FrameBodyWriter) WriteByte(b byte) error {
	return w.buf.WriteByte(b)
}

// WriteBytes writes a length-prefixed byte slice (int32 length).
// If data is nil, writes -1 as the length.
func (w *FrameBodyWriter) WriteBytes(data []byte) error {
	if data == nil {
		return binary.Write(&w.buf, binary.BigEndian, int32(-1))
	}
	if err := binary.Write(&w.buf, binary.BigEndian, int32(len(data))); err != nil {
		return err
	}
	_, err := w.buf.Write(data)
	return err
}

// WriteShortBytes writes a length-prefixed byte slice (uint16 length).
// If data is nil, writes 0xFFFF as the length.
func (w *FrameBodyWriter) WriteShortBytes(data []byte) error {
	if data == nil {
		return binary.Write(&w.buf, binary.BigEndian, uint16(0xFFFF))
	}
	if err := binary.Write(&w.buf, binary.BigEndian, uint16(len(data))); err != nil {
		return err
	}
	_, err := w.buf.Write(data)
	return err
}

// WriteString writes a length-prefixed string (uint16 length).
func (w *FrameBodyWriter) WriteString(s string) error {
	data := []byte(s)
	if err := binary.Write(&w.buf, binary.BigEndian, uint16(len(data))); err != nil {
		return err
	}
	_, err := w.buf.Write(data)
	return err
}

// WriteInt writes an int32 value.
func (w *FrameBodyWriter) WriteInt(v int32) error {
	return binary.Write(&w.buf, binary.BigEndian, v)
}

// WriteShort writes a uint16 value.
func (w *FrameBodyWriter) WriteShort(v uint16) error {
	return binary.Write(&w.buf, binary.BigEndian, v)
}

// WriteConsistency writes a consistency level.
func (w *FrameBodyWriter) WriteConsistency(consistency uint16) error {
	return w.WriteShort(consistency)
}

// WriteFlags writes protocol version-specific flags.
func (w *FrameBodyWriter) WriteFlags(flags uint32, protoVer byte) error {
	if protoVer >= ProtoVersion5 {
		return binary.Write(&w.buf, binary.BigEndian, flags)
	}
	return w.buf.WriteByte(byte(flags))
}

// Bytes returns the written bytes.
func (w *FrameBodyWriter) Bytes() []byte {
	return w.buf.Bytes()
}

// ============================================================================
// Logging Utilities
// ============================================================================

// FrameLogger provides structured logging for CQL frames.
type FrameLogger struct {
	logger *zap.Logger
}

// NewFrameLogger creates a new frame logger.
func NewFrameLogger(logger *zap.Logger) *FrameLogger {
	return &FrameLogger{logger: logger}
}

// LogReceivedFrame logs when a frame is received.
func (l *FrameLogger) LogReceivedFrame(frame *Frame) {
	l.logger.Debug("received frame",
		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)),
	)
}

// LogSendingFrame logs when a frame is being sent.
func (l *FrameLogger) LogSendingFrame(frame *Frame) {
	l.logger.Info("sending frame",
		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)),
	)
}

// LogProtocolVersion logs protocol version information.
func (l *FrameLogger) LogProtocolVersion(requestVersion, protoVer, responseVersion byte) {
	l.logger.Info("protocol version negotiation",
		zap.Uint8("request_version", requestVersion),
		zap.Uint8("protocol_version", protoVer),
		zap.Uint8("response_version", responseVersion),
	)
}

// ============================================================================
// Common Response Builders
// ============================================================================

// ResponseBuilder builds common CQL response frames.
type ResponseBuilder struct {
	version byte
}

// NewResponseBuilder creates a new response builder.
func NewResponseBuilder(version byte) *ResponseBuilder {
	return &ResponseBuilder{version: version}
}

// BuildReady creates a READY response frame.
func (b *ResponseBuilder) BuildReady(stream int16) *Frame {
	return &Frame{
		Version: 0,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeReady,
		Length:  0,
		Body:    nil,
	}
}

// BuildAuthenticate creates an AUTHENTICATE response frame.
func (b *ResponseBuilder) BuildAuthenticate(stream int16, authenticatorName string) *Frame {
	body := writeString(authenticatorName)
	return &Frame{
		Version: 0,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeAuthenticate,
		Length:  int32(len(body)),
		Body:    body,
	}
}

// BuildAuthSuccess creates an AUTH_SUCCESS response frame.
func (b *ResponseBuilder) BuildAuthSuccess(stream int16) *Frame {
	body := writeBytes([]byte{})
	return &Frame{
		Version: 0,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeAuthSuccess,
		Length:  int32(len(body)),
		Body:    body,
	}
}

// BuildSupported creates a SUPPORTED response frame.
func (b *ResponseBuilder) BuildSupported(stream int16, options map[string][]string) *Frame {
	body := encodeStringMultimap(options)
	return &Frame{
		Version: 0,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeSupported,
		Length:  int32(len(body)),
		Body:    body,
	}
}

// BuildVoidResult creates a VOID result frame.
func (b *ResponseBuilder) BuildVoidResult(stream int16) *Frame {
	body := make([]byte, 4)
	binary.BigEndian.PutUint32(body, uint32(ResultKindVoid))
	return &Frame{
		Version: 0,
		Flags:   0,
		Stream:  stream,
		Opcode:  OpcodeResult,
		Length:  int32(len(body)),
		Body:    body,
	}
}