//go:build integration
// +build integration

package integration_test

import (
	"context"
	"database/sql"
	"fmt"
	"net/url"
	"regexp"
	"strings"
	"testing"
	"time"

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

const genericRequestFailedMessage = "Request failed"

var integrationBackendErrorCodeRE = regexp.MustCompile(`\[[A-Za-z_]+:\d+\]\s*`)
var integrationBackendKeywordRE = regexp.MustCompile(`(?i)(^|[^[:alnum:]_'"\x60])(tidb|mysql)([^[:alnum:]_'"\x60]|$)`)
var integrationBackendWhitespaceRE = regexp.MustCompile(`\s+`)

func TestMySQLErrorPassthroughUsesSanitizedBackendMessage(t *testing.T) {
	requireProxy(t)

	session := newSession(t, sessionOptions{disableHostLookup: true})

	keyspace := uniqueKeyspace()
	if err := session.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		t.Fatalf("create keyspace failed: %v", err)
	}
	t.Cleanup(func() {
		_ = session.Query(dropKeyspaceCQL(keyspace)).Exec()
	})

	db, err := sql.Open("mysql", mysqlDSNForDatabase(mysqlCfg, keyspace))
	if err != nil {
		t.Fatalf("mysql connect failed: %v", err)
	}
	defer db.Close()

	t.Logf("backend version: %s", fetchBackendVersion(t, db))

	table := "fn_passthrough"
	create := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY, name text, idx int)", keyspace, table)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	if err := session.Query(fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (1, 'seed', 7)", keyspace, table)).Exec(); err != nil {
		t.Fatalf("seed row failed: %v", err)
	}

	cases := []struct {
		name     string
		mysqlSQL string
		cqlQuery string
	}{
		{
			name:     "select-invalid-arg-count-abs",
			mysqlSQL: "SELECT ABS(idx, idx) FROM `fn_passthrough` WHERE id = 1",
			cqlQuery: fmt.Sprintf("SELECT abs(idx, idx) FROM %s.%s WHERE id = 1", keyspace, table),
		},
		{
			name:     "select-invalid-arg-count-power",
			mysqlSQL: "SELECT POWER(idx) FROM `fn_passthrough` WHERE id = 1",
			cqlQuery: fmt.Sprintf("SELECT power(idx) FROM %s.%s WHERE id = 1", keyspace, table),
		},
		{
			name:     "insert-invalid-arg-count-abs",
			mysqlSQL: "INSERT INTO `fn_passthrough` (id, name, idx) VALUES (3, 'gamma', ABS(1, 2))",
			cqlQuery: fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (3, 'gamma', abs(1, 2))", keyspace, table),
		},
		{
			name:     "insert-invalid-arg-count-power",
			mysqlSQL: "INSERT INTO `fn_passthrough` (id, name, idx) VALUES (4, 'delta', POWER(2))",
			cqlQuery: fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (4, 'delta', power(2))", keyspace, table),
		},
	}

	for i, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			mysqlErr := runMySQLError(t, db, tc.mysqlSQL)
			want := expectedSanitizedPassthroughMessage(mysqlErr)

			cqlErr := session.Query(tc.cqlQuery).Exec()
			logErrorComparison(t, tc.name, mysqlErr, cqlErr)
			requireErrorCode(t, cqlErr, cql.ErrorCodeInvalid)
			requireCQLMessage(t, cqlErr, want)
			requirePassthroughMessageSanitized(t, mysqlMessage(mysqlErr), cqlMessage(cqlErr))

			client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
			resp := executeRawQuery(t, client, tc.cqlQuery, int16(i+1))
			code, msg := parseError(t, resp)
			if code != cql.ErrorCodeInvalid {
				t.Fatalf("expected ErrorCodeInvalid, got %d", code)
			}
			if msg != want {
				t.Fatalf("unexpected raw protocol error message: got %q want %q", msg, want)
			}
			requirePassthroughMessageSanitized(t, mysqlMessage(mysqlErr), msg)
		})
	}
}

func fetchBackendVersion(t *testing.T, db *sql.DB) string {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	var version string
	if err := db.QueryRowContext(ctx, "SELECT VERSION()").Scan(&version); err != nil {
		t.Fatalf("fetch backend version failed: %v", err)
	}
	return version
}

func mysqlDSNForDatabase(cfg mysqlConfig, schema string) string {
	params := url.Values{}
	params.Add("parseTime", "true")
	params.Add("loc", "UTC")
	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?%s",
		url.QueryEscape(cfg.user),
		url.QueryEscape(cfg.password),
		cfg.host,
		cfg.port,
		url.PathEscape(schema),
		params.Encode(),
	)
}

func runMySQLError(t *testing.T, db *sql.DB, stmt string) error {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if strings.HasPrefix(strings.ToUpper(strings.TrimSpace(stmt)), "SELECT") {
		rows, err := db.QueryContext(ctx, stmt)
		if rows != nil {
			defer rows.Close()
		}
		if err == nil {
			t.Fatalf("expected mysql error for %q, got nil", stmt)
		}
		return err
	}

	_, err := db.ExecContext(ctx, stmt)
	if err == nil {
		t.Fatalf("expected mysql error for %q, got nil", stmt)
	}
	return err
}

func expectedSanitizedPassthroughMessage(err error) string {
	detail := sanitizeBackendMessageForExpectation(mysqlMessage(err))
	if detail == "" {
		return genericRequestFailedMessage
	}
	return genericRequestFailedMessage + ": " + detail
}

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

	sanitized = integrationBackendErrorCodeRE.ReplaceAllString(sanitized, "")
	sanitized = integrationBackendKeywordRE.ReplaceAllString(sanitized, "${1}backend${3}")
	sanitized = integrationBackendWhitespaceRE.ReplaceAllString(sanitized, " ")
	return strings.TrimSpace(sanitized)
}

func requirePassthroughMessageSanitized(t *testing.T, backendMsg, got string) {
	t.Helper()
	if !strings.HasPrefix(got, genericRequestFailedMessage) {
		t.Fatalf("expected passthrough message to start with %q, got %q", genericRequestFailedMessage, got)
	}
	if integrationBackendErrorCodeRE.MatchString(got) {
		t.Fatalf("expected protocol/backend error code prefix to be stripped, got %q", got)
	}

	backendLower := strings.ToLower(backendMsg)
	gotLower := strings.ToLower(got)
	if strings.Contains(backendLower, "tidb") && strings.Contains(gotLower, "tidb") {
		t.Fatalf("expected TiDB keyword to be desensitized, got %q", got)
	}
	if strings.Contains(backendLower, "mysql") && strings.Contains(gotLower, "mysql") {
		t.Fatalf("expected MySQL keyword to be desensitized, got %q", got)
	}
}