package executor

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"strings"

	"github.com/go-sql-driver/mysql"
	"go.uber.org/zap"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/cqlerrors"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

type mysqlOperation uint8

const (
	mysqlOpUnknown mysqlOperation = iota
	mysqlOpRead
	mysqlOpWrite
)

const (
	mysqlErrDBExists        = 1007
	mysqlErrBadDB           = 1049
	mysqlErrTableExists     = 1050
	mysqlErrDuplicateColumn = 1060
	mysqlErrDuplicateIndex  = 1061
	mysqlErrDuplicateKey    = 1062
	mysqlErrBlobTextKey     = 1170
	mysqlErrNoSuchTable     = 1146
	mysqlErrLockWaitTimeout = 1205
	mysqlErrLockDeadlock    = 1213
)

const (
	mysqlGenericMessage           = "Request failed"
	mysqlKeyspaceExistsMessage    = "Keyspace already exists"
	mysqlTableExistsMessage       = "Table already exists"
	mysqlDuplicateKeyMessage      = "Duplicate key"
	mysqlDuplicateIndexMessage    = "Duplicate index"
	mysqlDuplicateColumnMessage   = "Duplicate column"
	mysqlKeyspaceNotFoundMessage  = "Keyspace does not exist"
	mysqlTableNotFoundMessage     = "Table does not exist"
	mysqlBlobTextIndexMessage     = "Index on BLOB/TEXT column is not supported"
	mysqlReadTimeoutMessage       = "Read timeout"
	mysqlWriteTimeoutMessage      = "Write timeout"
	mysqlRequestTimedOutMessage   = "Request timed out"
	mysqlBackendConnectionMessage = "Backend connection error"
)

func (op mysqlOperation) String() string {
	switch op {
	case mysqlOpRead:
		return "read"
	case mysqlOpWrite:
		return "write"
	default:
		return "unknown"
	}
}

func (e *Executor) mapMySQLError(ctx context.Context, err error, op mysqlOperation) error {
	return mapMySQLErrorWithLogger(ctx, err, op, e.logger)
}

func (e *Executor) mapMySQLErrorWithSQL(ctx context.Context, err error, op mysqlOperation, sql string) error {
	return mapMySQLErrorWithLoggerAndSQL(ctx, err, op, e.logger, sql)
}

func mapMySQLError(err error, op mysqlOperation) error {
	return mapMySQLErrorWithLogger(context.Background(), err, op, nil)
}

func mapMySQLErrorWithLogger(ctx context.Context, err error, op mysqlOperation, logger *zap.Logger) error {
	return mapMySQLErrorWithLoggerAndSQL(ctx, err, op, logger, "")
}

func mapMySQLErrorWithLoggerAndSQL(ctx context.Context, err error, op mysqlOperation, logger *zap.Logger, sql string) error {
	if err == nil {
		return nil
	}
	errType := metrics.MySQLErrorOther
	fields := []zap.Field{zap.String("op", op.String())}
	if sql != "" {
		fields = append(fields, zap.String("sql", sql))
	}
	if errors.Is(err, context.DeadlineExceeded) {
		errType = metrics.MySQLErrorTimeout
		metrics.RecordMySQLError(ctx, op.String(), errType)
		if logger != nil {
			fields = append(fields, zap.Error(err))
			logger.Warn("mysql request timed out", fields...)
		}
		if mapped := timeoutError(op, timeoutMessage(op)); mapped != nil {
			return mapped
		}
		return cqlerrors.Invalid(mysqlGenericMessage)
	}
	var myErr *mysql.MySQLError
	if errors.As(err, &myErr) {
		errType = metrics.MySQLErrorDriver
		metrics.RecordMySQLError(ctx, op.String(), errType)
		if logger != nil {
			fields = append(fields,
				zap.Uint16("mysql_code", myErr.Number),
				zap.String("mysql_message", myErr.Message),
				zap.Error(err),
			)
			logger.Warn("mysql error", fields...)
		}
		switch myErr.Number {
		case mysqlErrDBExists:
			return cqlerrors.AlreadyExists(mysqlKeyspaceExistsMessage)
		case mysqlErrTableExists:
			return cqlerrors.AlreadyExists(mysqlTableExistsMessage)
		case mysqlErrDuplicateKey:
			return cqlerrors.AlreadyExists(mysqlDuplicateKeyMessage)
		case mysqlErrDuplicateIndex:
			return cqlerrors.AlreadyExists(mysqlDuplicateIndexMessage)
		case mysqlErrDuplicateColumn:
			return cqlerrors.AlreadyExists(mysqlDuplicateColumnMessage)
		case mysqlErrBadDB:
			return cqlerrors.Invalid(mysqlKeyspaceNotFoundMessage)
		case mysqlErrNoSuchTable:
			return cqlerrors.Invalid(mysqlTableNotFoundMessage)
		case mysqlErrBlobTextKey:
			return cqlerrors.Invalid(blobTextIndexMessage(myErr.Message))
		case mysqlErrLockWaitTimeout, mysqlErrLockDeadlock:
			// A deadlock fully rolls back the transaction, so a write is safe to
			// retry transparently; tag it so the retry layer can pick it up.
			if op == mysqlOpWrite && myErr.Number == mysqlErrLockDeadlock {
				return cqlerrors.WriteTimeoutRetryable(timeoutMessage(op), cqlerrors.RetryDeadlock)
			}
			if mapped := timeoutError(op, timeoutMessage(op)); mapped != nil {
				return mapped
			}
			return cqlerrors.Invalid(mysqlGenericMessage)
		}
		return cqlerrors.Invalid(mysqlGenericMessageWithBackendDetail(myErr.Message))
	}
	if logger != nil && mysqlbackend.IsMySQLConnectionError(err) {
		errType = metrics.MySQLErrorConnection
		metrics.RecordMySQLError(ctx, op.String(), errType)
		fields = append(fields, zap.Error(err))
		logger.Warn("mysql connection error", fields...)
		return cqlerrors.Unavailable(mysqlBackendConnectionMessage)
	}
	metrics.RecordMySQLError(ctx, op.String(), errType)
	return err
}

func operationLabelForQuery(op mysqlOperation) string {
	if op == mysqlOpWrite {
		return metrics.MySQLOpQueryWrite
	}
	return metrics.MySQLOpQueryRead
}

func timeoutError(op mysqlOperation, msg string) error {
	switch op {
	case mysqlOpRead:
		return cqlerrors.ReadTimeout(msg)
	case mysqlOpWrite:
		return cqlerrors.WriteTimeout(msg)
	default:
		return nil
	}
}

func timeoutMessage(op mysqlOperation) string {
	switch op {
	case mysqlOpRead:
		return mysqlReadTimeoutMessage
	case mysqlOpWrite:
		return mysqlWriteTimeoutMessage
	default:
		return mysqlRequestTimedOutMessage
	}
}

var blobTextColumnRE = regexp.MustCompile(`column\\s+'([^']+)'`)
var backendErrorCodeRE = regexp.MustCompile(`\[[A-Za-z_]+:\d+\]\s*`)
var backendErrorPrefixRE = regexp.MustCompile(`(?i)(^|[:;]\s*)(?:error|errno)\s+\d+(?:\s*\([0-9A-Z]+\))?:\s*`)
var backendKeywordRE = regexp.MustCompile(`(?i)(^|[^[:alnum:]_'"\x60])(tidb|mysql|tikv)([^[:alnum:]_'"\x60]|$)`)
var backendWhitespaceRE = regexp.MustCompile(`\s+`)
var translateCQLToSQLPrefixRE = regexp.MustCompile(`(?i)^failed to translate cql to sql:\s*`)

func blobTextIndexMessage(mysqlMessage string) string {
	matches := blobTextColumnRE.FindStringSubmatch(mysqlMessage)
	if len(matches) == 2 {
		return fmt.Sprintf("Index on BLOB/TEXT column %q is not supported", matches[1])
	}
	return mysqlBlobTextIndexMessage
}

func mysqlGenericMessageWithBackendDetail(msg string) string {
	detail := sanitizeBackendErrorMessage(msg)
	if detail == "" {
		return mysqlGenericMessage
	}
	return fmt.Sprintf("%s: %s", mysqlGenericMessage, detail)
}

func clientVisibleRequestError(err error) error {
	if err == nil {
		return nil
	}
	var coder interface {
		error
		Code() int
	}
	if errors.As(err, &coder) {
		return coder
	}
	detail := sanitizeClientVisibleMessage(err.Error())
	if detail == "" {
		detail = mysqlGenericMessage
	}
	return cqlerrors.Invalid(detail)
}

func sanitizeClientVisibleMessage(msg string) string {
	sanitized := strings.TrimSpace(msg)
	if sanitized == "" {
		return ""
	}
	sanitized = translateCQLToSQLPrefixRE.ReplaceAllString(sanitized, "")
	sanitized = sanitizeBackendErrorMessage(sanitized)
	return strings.TrimSpace(sanitized)
}

func sanitizeBackendErrorMessage(msg string) string {
	sanitized := strings.TrimSpace(msg)
	if sanitized == "" {
		return ""
	}

	sanitized = backendErrorPrefixRE.ReplaceAllString(sanitized, "${1}")
	sanitized = backendErrorCodeRE.ReplaceAllString(sanitized, "")
	sanitized = backendKeywordRE.ReplaceAllString(sanitized, "${1}backend${3}")
	sanitized = backendWhitespaceRE.ReplaceAllString(sanitized, " ")
	sanitized = strings.TrimSpace(sanitized)
	sanitized = strings.TrimLeft(sanitized, ":;- ")
	return strings.TrimSpace(sanitized)
}