//go:build integration
// +build integration

package integration_test

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

	gocql "github.com/apache/cassandra-gocql-driver/v2"
	_ "github.com/go-sql-driver/mysql"
	"go.uber.org/zap"

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

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

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

	host, port, stop := startTestProxy(t, mysqlCfg)
	t.Cleanup(stop)

	session := newSessionForServer(t, host, port)

	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() })

	table := "lock_wait"
	if err := session.Query(createTableCQL(keyspace, table)).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, 'a', 1)",
		keyspace,
		table,
	)).Exec(); err != nil {
		t.Fatalf("insert row failed: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	lockConn, err := db.Conn(ctx)
	if err != nil {
		t.Fatalf("get lock connection failed: %v", err)
	}
	defer lockConn.Close()

	mustExecSessionSQL(t, ctx, lockConn, "SET SESSION tidb_txn_mode = 'pessimistic'")
	mustExecSessionSQL(t, ctx, lockConn, "SET SESSION innodb_lock_wait_timeout = 1")

	tx, err := lockConn.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin tx failed: %v", err)
	}
	defer func() {
		_ = tx.Rollback()
	}()

	lockSQL := fmt.Sprintf("SELECT id FROM `%s`.`%s` WHERE id = 1 FOR UPDATE", keyspace, table)
	var lockedID int
	if err := tx.QueryRowContext(ctx, lockSQL).Scan(&lockedID); err != nil {
		t.Fatalf("lock row failed: %v", err)
	}
	if lockedID != 1 {
		t.Fatalf("unexpected locked row id: %d", lockedID)
	}

	batch := session.NewBatch(gocql.LoggedBatch)
	batch.Query(fmt.Sprintf(
		"UPDATE %s.%s SET idx = 2 WHERE id = 1",
		keyspace,
		table,
	))

	err = session.ExecuteBatch(batch)
	requireErrorCode(t, err, cql.ErrorCodeWriteTimeout)
}

func mustExecSessionSQL(t *testing.T, ctx context.Context, conn *sql.Conn, stmt string) {
	t.Helper()
	if _, err := conn.ExecContext(ctx, stmt); err != nil {
		t.Fatalf("exec session SQL %q failed: %v", stmt, err)
	}
}

func startTestProxy(t *testing.T, cfg mysqlConfig) (string, int, func()) {
	t.Helper()
	log := zap.NewNop()

	poolCfg := &mysqlbackend.PoolConfig{
		Host:            fmt.Sprintf("%s:%d", cfg.host, cfg.port),
		MaxOpenConns:    10,
		MaxIdleConns:    5,
		ConnectTimeout:  5 * time.Second,
		ReadTimeout:     15 * time.Second,
		WriteTimeout:    15 * time.Second,
		ConnMaxLifetime: 2 * time.Minute,
		ConnMaxIdleTime: 30 * time.Second,
		DSNParams: map[string]string{
			"innodb_lock_wait_timeout": "1",
			"tidb_txn_mode":            "'pessimistic'",
		},
	}

	poolMgr := mysqlbackend.NewPoolManager(poolCfg, log)
	assertProxySessionVariables(t, poolMgr, cfg)
	queryExecutor := executor.NewExecutor(poolMgr, log)
	queryExecutor.Start()
	authenticator := cql.NewPasswordAuthenticator()
	handler := cql.NewMessageHandlerWithExecutor(log, authenticator, queryExecutor)

	server := cql.NewServer("127.0.0.1:0", handler, log, &cql.ServerOptions{
		ProtocolVersion: cql.ProtoVersion4,
	})
	if err := server.Start(); err != nil {
		_ = poolMgr.Close()
		t.Fatalf("start test server failed: %v", err)
	}

	host, port := parseTCPAddr(server.Addr())
	stop := func() {
		_ = server.Stop()
		queryExecutor.Close()
		_ = poolMgr.Close()
	}
	return host, port, stop
}

func assertProxySessionVariables(t *testing.T, poolMgr *mysqlbackend.PoolManager, cfg mysqlConfig) {
	t.Helper()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	pool, err := poolMgr.GetPool(cfg.user, cfg.password)
	if err != nil {
		t.Fatalf("get proxy pool failed: %v", err)
	}

	var (
		lockWait int
		txnMode  string
	)
	if err := pool.QueryRow(ctx, "SELECT @@SESSION.innodb_lock_wait_timeout, LOWER(@@SESSION.tidb_txn_mode)").Scan(&lockWait, &txnMode); err != nil {
		t.Fatalf("query proxy session variables failed: %v", err)
	}
	if lockWait != 1 {
		t.Fatalf("unexpected proxy session innodb_lock_wait_timeout: %d", lockWait)
	}
	if txnMode != "pessimistic" {
		t.Fatalf("unexpected proxy session tidb_txn_mode: %q", txnMode)
	}
}

func newSessionForServer(t *testing.T, host string, port int) *gocql.Session {
	t.Helper()

	cluster := gocql.NewCluster(host)
	cluster.Port = port
	cluster.ProtoVersion = 4
	cluster.NumConns = 1
	cluster.ConnectTimeout = 5 * time.Second
	cluster.Timeout = 10 * time.Second
	cluster.DisableInitialHostLookup = true
	cluster.Authenticator = gocql.PasswordAuthenticator{
		Username:              mysqlCfg.user,
		Password:              mysqlCfg.password,
		AllowedAuthenticators: []string{"org.apache.cassandra.auth.PasswordAuthenticator"},
	}

	session, err := cluster.CreateSession()
	if err != nil {
		t.Fatalf("create test session failed: %v", err)
	}
	t.Cleanup(func() { session.Close() })
	return session
}