package executor

import (
	"context"
	"errors"
	"fmt"
	"testing"

	"github.com/go-sql-driver/mysql"

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

func requireCQLCode(t *testing.T, err error, code int) {
	t.Helper()
	if err == nil {
		t.Fatalf("expected error code %d, got nil", code)
	}
	var coder interface{ Code() int }
	if !errors.As(err, &coder) {
		t.Fatalf("expected cql error code %d, got %T: %v", code, err, err)
	}
	if got := coder.Code(); got != code {
		t.Fatalf("expected error code %d, got %d", code, got)
	}
}

func TestMapMySQLErrorDuplicateEntry(t *testing.T) {
	err := mapMySQLError(&mysql.MySQLError{
		Number:  mysqlErrDuplicateKey,
		Message: "duplicate entry",
	}, mysqlOpWrite)
	requireCQLCode(t, err, cqlerrors.CodeAlreadyExists)
}

func TestMapMySQLErrorInvalid(t *testing.T) {
	err := mapMySQLError(&mysql.MySQLError{
		Number:  mysqlErrBadDB,
		Message: "unknown database",
	}, mysqlOpRead)
	requireCQLCode(t, err, cqlerrors.CodeInvalid)
}

func TestMapMySQLErrorTimeouts(t *testing.T) {
	cases := []struct {
		name string
		err  error
		op   mysqlOperation
		code int
	}{
		{
			name: "read-deadline",
			err:  fmt.Errorf("wrapped: %w", context.DeadlineExceeded),
			op:   mysqlOpRead,
			code: cqlerrors.CodeReadTimeout,
		},
		{
			name: "write-deadline",
			err:  context.DeadlineExceeded,
			op:   mysqlOpWrite,
			code: cqlerrors.CodeWriteTimeout,
		},
		{
			name: "read-lock-wait",
			err: &mysql.MySQLError{
				Number:  mysqlErrLockWaitTimeout,
				Message: "lock wait timeout",
			},
			op:   mysqlOpRead,
			code: cqlerrors.CodeReadTimeout,
		},
		{
			name: "write-lock-wait",
			err: &mysql.MySQLError{
				Number:  mysqlErrLockWaitTimeout,
				Message: "lock wait timeout",
			},
			op:   mysqlOpWrite,
			code: cqlerrors.CodeWriteTimeout,
		},
		{
			name: "read-deadlock",
			err: &mysql.MySQLError{
				Number:  mysqlErrLockDeadlock,
				Message: "deadlock found",
			},
			op:   mysqlOpRead,
			code: cqlerrors.CodeReadTimeout,
		},
		{
			name: "write-deadlock",
			err: &mysql.MySQLError{
				Number:  mysqlErrLockDeadlock,
				Message: "deadlock found",
			},
			op:   mysqlOpWrite,
			code: cqlerrors.CodeWriteTimeout,
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := mapMySQLError(tc.err, tc.op)
			requireCQLCode(t, err, tc.code)
		})
	}
}

func TestMapMySQLErrorDeadlockRetryKind(t *testing.T) {
	cases := []struct {
		name     string
		number   uint16
		op       mysqlOperation
		wantCode int
		wantKind cqlerrors.RetryKind
	}{
		{"write-deadlock-retryable", mysqlErrLockDeadlock, mysqlOpWrite, cqlerrors.CodeWriteTimeout, cqlerrors.RetryDeadlock},
		{"write-lock-wait-not-retryable", mysqlErrLockWaitTimeout, mysqlOpWrite, cqlerrors.CodeWriteTimeout, cqlerrors.RetryNone},
		{"read-deadlock-not-retryable", mysqlErrLockDeadlock, mysqlOpRead, cqlerrors.CodeReadTimeout, cqlerrors.RetryNone},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := mapMySQLError(&mysql.MySQLError{Number: tc.number, Message: "x"}, tc.op)
			requireCQLCode(t, err, tc.wantCode)
			if got := cqlerrors.RetryKindOf(err); got != tc.wantKind {
				t.Fatalf("expected retry kind %v, got %v", tc.wantKind, got)
			}
		})
	}
}

func TestMapMySQLErrorUnknownUsesGenericMessage(t *testing.T) {
	err := mapMySQLError(&mysql.MySQLError{
		Number:  1105,
		Message: "",
	}, mysqlOpWrite)
	requireCQLCode(t, err, cqlerrors.CodeInvalid)
	if err == nil || err.Error() != mysqlGenericMessage {
		t.Fatalf("expected generic mysql message, got %v", err)
	}
}

func TestMapMySQLErrorUnknownPassesThroughSanitizedBackendMessage(t *testing.T) {
	cases := []struct {
		name    string
		message string
		want    string
	}{
		{
			name:    "tidb planner prefix",
			message: "[planner:1054]Unknown column 'c1' in 'field list'",
			want:    "Request failed: Unknown column 'c1' in 'field list'",
		},
		{
			name:    "tidb type conversion prefix",
			message: "[types:1292]Truncated incorrect DOUBLE value: 'hello'",
			want:    "Request failed: Truncated incorrect DOUBLE value: 'hello'",
		},
		{
			name:    "tidb ddl prefix",
			message: "[ddl:8200]Unsupported local temp table partition",
			want:    "Request failed: Unsupported local temp table partition",
		},
		{
			name:    "tidb keyword redacted",
			message: "[ddl:8244]create binding for TiDB internal SQL is unsupported",
			want:    "Request failed: create binding for backend internal SQL is unsupported",
		},
		{
			name:    "mysql keyword redacted",
			message: "The MySQL server is running with the --read-only option so it cannot execute this statement",
			want:    "Request failed: The backend server is running with the --read-only option so it cannot execute this statement",
		},
		{
			name:    "numeric prefix redacted",
			message: "Error 1105 (HY000): [planner:1054]Unknown column 'c1' in 'field list'",
			want:    "Request failed: Unknown column 'c1' in 'field list'",
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := mapMySQLError(&mysql.MySQLError{
				Number:  1105,
				Message: tc.message,
			}, mysqlOpWrite)
			requireCQLCode(t, err, cqlerrors.CodeInvalid)
			if err == nil || err.Error() != tc.want {
				t.Fatalf("unexpected message: got %v want %q", err, tc.want)
			}
		})
	}
}

func TestMapMySQLErrorBlobTextIndex(t *testing.T) {
	err := mapMySQLError(&mysql.MySQLError{
		Number:  mysqlErrBlobTextKey,
		Message: "BLOB/TEXT column 'c' used in key specification without a key length",
	}, mysqlOpWrite)
	requireCQLCode(t, err, cqlerrors.CodeInvalid)
	if err == nil || err.Error() != "Index on BLOB/TEXT column is not supported" {
		t.Fatalf("unexpected message: %v", err)
	}
}

func TestClientVisibleRequestErrorStripsTranslatePrefix(t *testing.T) {
	err := clientVisibleRequestError(fmt.Errorf("failed to translate CQL to SQL: %w", errors.New("unsupported WHERE conditions require ALLOW FILTERING")))
	requireCQLCode(t, err, cqlerrors.CodeInvalid)
	if err == nil || err.Error() != "unsupported WHERE conditions require ALLOW FILTERING" {
		t.Fatalf("unexpected sanitized message: %v", err)
	}
}

func TestClientVisibleRequestErrorPreservesCQLCode(t *testing.T) {
	src := cqlerrors.Invalid("query requires ALLOW FILTERING")
	err := clientVisibleRequestError(fmt.Errorf("wrapped: %w", src))
	requireCQLCode(t, err, cqlerrors.CodeInvalid)
	if err == nil || err.Error() != src.Error() {
		t.Fatalf("expected wrapped cql error to be preserved, got %v", err)
	}
}