//go:build integration
// +build integration

package integration_test

import (
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"

	gocql "github.com/apache/cassandra-gocql-driver/v2"

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

// TestBatchDeadlockIsNotRetried verifies that batch requests are excluded from
// deadlock retry. Two batches lock the same rows in opposite order, which makes
// TiDB raise a deadlock (MySQL 1213) on one of them; because batches are not
// retried, the loser surfaces a WriteTimeout to the client and the retry
// counter does not advance. (Single-statement deadlock retry is covered by the
// executor unit tests.)
func TestBatchDeadlockIsNotRetried(t *testing.T) {
	requireProxy(t)

	// Dedicated proxy with metrics enabled (to read the retry counter) and
	// generous timeouts (so non-deadlock waits resolve instead of timing out).
	// Retry is on by default; the batch exclusion is what we're asserting.
	retryEnabled := true
	host, port, stop, err := startProxyWithConfig(mysqlCfg, func(c *config.Config) {
		c.Server.Admin.Metrics.Enabled = true
		c.MySQL.ReadTimeout = 30 * time.Second
		c.MySQL.WriteTimeout = 30 * time.Second
		c.MySQL.Retry.Deadlock.Enabled = &retryEnabled
	})
	if err != nil {
		t.Fatalf("start proxy failed: %v", err)
	}
	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 := "deadlock"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	const rows = 10
	for id := 1; id <= rows; id++ {
		if err := session.Query(fmt.Sprintf(
			"INSERT INTO %s.%s (id, name, idx) VALUES (%d, 'seed', 0)",
			keyspace, table, id,
		)).Exec(); err != nil {
			t.Fatalf("seed row %d failed: %v", id, err)
		}
	}

	updateInOrder := func(ids []int, name string) error {
		batch := session.NewBatch(gocql.LoggedBatch)
		for _, id := range ids {
			batch.Query(fmt.Sprintf(
				"UPDATE %s.%s SET name = '%s' WHERE id = %d",
				keyspace, table, name, id,
			))
		}
		return session.ExecuteBatch(batch)
	}

	asc := make([]int, rows)
	desc := make([]int, rows)
	for i := 0; i < rows; i++ {
		asc[i] = i + 1
		desc[i] = rows - i
	}

	before := metrics.WriteRetryTotal()

	const rounds = 50
	// Per-round rendezvous: buffered (cap 1) so each party posts its token
	// before waiting on the other's (an unbuffered pair would deadlock the
	// test itself).
	ascReady := make(chan struct{}, 1)
	descReady := make(chan struct{}, 1)
	errCh := make(chan error, 2*rounds)

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		for i := 0; i < rounds; i++ {
			ascReady <- struct{}{}
			<-descReady
			if err := updateInOrder(asc, "a"); err != nil {
				errCh <- err
			}
		}
	}()
	go func() {
		defer wg.Done()
		for i := 0; i < rounds; i++ {
			descReady <- struct{}{}
			<-ascReady
			if err := updateInOrder(desc, "b"); err != nil {
				errCh <- err
			}
		}
	}()
	wg.Wait()
	close(errCh)

	writeTimeouts := 0
	var others []error
	for err := range errCh {
		var reqErr gocql.RequestError
		if errors.As(err, &reqErr) && reqErr.Code() == cql.ErrorCodeWriteTimeout {
			writeTimeouts++
			continue
		}
		others = append(others, err)
	}

	if writeTimeouts == 0 {
		t.Fatalf("expected at least one batch to surface a deadlock WriteTimeout (batches must not be retried); other errors: %v", others)
	}
	if delta := metrics.WriteRetryTotal() - before; delta != 0 {
		t.Fatalf("batch requests must not be deadlock-retried, but the retry counter advanced by %.0f", delta)
	}
	t.Logf("batch deadlocks surfaced without retry: %d WriteTimeout errors over %d rounds", writeTimeouts, rounds)
}