//go:build integration
// +build integration

package integration_test

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	gocql "github.com/apache/cassandra-gocql-driver/v2"
	mysql "github.com/go-sql-driver/mysql"

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

func TestMySQLErrorSanitization(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", mysqlDSN(mysqlCfg))
	if err != nil {
		t.Fatalf("mysql connect failed: %v", err)
	}
	defer db.Close()

	t.Run("keyspace-exists", func(t *testing.T) {
		mysqlErr := execMySQLError(t, db, fmt.Sprintf("CREATE DATABASE `%s`", keyspace))
		cqlErr := session.Query(createKeyspaceCQL(keyspace)).Exec()
		logErrorComparison(t, "keyspace-exists", mysqlErr, cqlErr)
		requireErrorCode(t, cqlErr, cql.ErrorCodeAlreadyExists)
		requireCQLMessage(t, cqlErr, "Keyspace already exists")
	})

	t.Run("table-exists", func(t *testing.T) {
		table := "tbl_exists"
		create := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY)", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		mysqlErr := execMySQLError(t, db, fmt.Sprintf("CREATE TABLE `%s`.`%s` (id INT PRIMARY KEY)", keyspace, table))
		cqlErr := execCQLRetryingInformationSchemaChanged(t, session, create)
		logErrorComparison(t, "table-exists", mysqlErr, cqlErr)
		requireErrorCode(t, cqlErr, cql.ErrorCodeAlreadyExists)
		requireCQLMessage(t, cqlErr, "Table already exists")
	})

	t.Run("duplicate-column", func(t *testing.T) {
		table := "dup_col"
		create := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY)", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		if err := session.Query(fmt.Sprintf("ALTER TABLE %s.%s ADD c int", keyspace, table)).Exec(); err != nil {
			t.Fatalf("alter table add column failed: %v", err)
		}
		mysqlErr := execMySQLError(t, db, fmt.Sprintf("ALTER TABLE `%s`.`%s` ADD COLUMN c INT", keyspace, table))
		cqlErr := session.Query(fmt.Sprintf("ALTER TABLE %s.%s ADD c int", keyspace, table)).Exec()
		logErrorComparison(t, "duplicate-column", mysqlErr, cqlErr)
		requireErrorCode(t, cqlErr, cql.ErrorCodeAlreadyExists)
		requireCQLMessage(t, cqlErr, "Duplicate column")
	})

	t.Run("duplicate-index", func(t *testing.T) {
		table := "dup_idx"
		create := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY, c int)", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		if err := session.Query(fmt.Sprintf("CREATE INDEX idx_c ON %s.%s (c)", keyspace, table)).Exec(); err != nil {
			t.Fatalf("create index failed: %v", err)
		}
		mysqlErr := execMySQLError(t, db, fmt.Sprintf("CREATE INDEX idx_c ON `%s`.`%s` (c)", keyspace, table))
		cqlErr := session.Query(fmt.Sprintf("CREATE INDEX idx_c ON %s.%s (c)", keyspace, table)).Exec()
		logErrorComparison(t, "duplicate-index", mysqlErr, cqlErr)
		requireErrorCode(t, cqlErr, cql.ErrorCodeAlreadyExists)
		requireCQLMessage(t, cqlErr, "Duplicate index")
	})

	t.Run("blob-text-index-auto-prefix", func(t *testing.T) {
		table := "text_idx"
		create := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY, c text)", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		mysqlErr := execMySQLError(t, db, fmt.Sprintf("CREATE INDEX idx_c ON `%s`.`%s` (c)", keyspace, table))
		cqlErr := session.Query(fmt.Sprintf("CREATE INDEX idx_c ON %s.%s (c)", keyspace, table)).Exec()
		logErrorComparison(t, "blob-text-index", mysqlErr, cqlErr)
		if cqlErr != nil {
			t.Fatalf("expected proxy to create prefixed TEXT index, got: %v", cqlErr)
		}
	})
}

func execMySQLError(t *testing.T, db *sql.DB, stmt string) error {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_, err := db.ExecContext(ctx, stmt)
	if err == nil {
		t.Fatalf("expected mysql error for %q, got nil", stmt)
	}
	return err
}

func execCQLRetryingInformationSchemaChanged(t *testing.T, session *gocql.Session, stmt string) error {
	t.Helper()
	var err error
	for attempt := 0; attempt < 10; attempt++ {
		err = session.Query(stmt).Exec()
		if !isInformationSchemaChanged(err) {
			return err
		}
		time.Sleep(100 * time.Millisecond)
	}
	return err
}

func isInformationSchemaChanged(err error) bool {
	var reqErr gocql.RequestError
	return errors.As(err, &reqErr) && strings.Contains(reqErr.Message(), "Information schema is changed")
}

func requireCQLMessage(t *testing.T, err error, expected string) {
	t.Helper()
	if err == nil {
		t.Fatalf("expected cql error, got nil")
	}
	var reqErr gocql.RequestError
	if !errors.As(err, &reqErr) {
		t.Fatalf("expected RequestError, got %T: %v", err, err)
	}
	if reqErr.Message() != expected {
		t.Fatalf("expected message %q, got %q", expected, reqErr.Message())
	}
}

func mysqlMessage(err error) string {
	var myErr *mysql.MySQLError
	if errors.As(err, &myErr) {
		return myErr.Message
	}
	if err == nil {
		return ""
	}
	return err.Error()
}

func cqlMessage(err error) string {
	var reqErr gocql.RequestError
	if errors.As(err, &reqErr) {
		return reqErr.Message()
	}
	if err == nil {
		return ""
	}
	return err.Error()
}

func logErrorComparison(t *testing.T, name string, mysqlErr, cqlErr error) {
	t.Helper()
	t.Logf("[%s] mysql: %s", name, mysqlMessage(mysqlErr))
	t.Logf("[%s] cql:   %s", name, cqlMessage(cqlErr))
}