//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"
	"time"

	"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/mysqlbackend"
)

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

	prevCfg := config.GetConfig()
	host, port, stop, poolMgr, err := startProxyWithConfigAndPool(mysqlCfg, func(cfg *config.Config) {
		cfg.MySQL.MaxOpenConns = 3
		cfg.MySQL.MaxIdleConns = 3
		cfg.Executor.UnloggedBatchParallel = true
		cfg.Executor.UnloggedBatchParallelism = 4
		cfg.Executor.UnloggedBatchMaxMySQLConnections = 2
	})
	if err != nil {
		config.SetConfig(prevCfg)
		t.Fatalf("start proxy with select headroom config failed: %v", err)
	}
	defer func() {
		stop()
		config.SetConfig(prevCfg)
	}()

	batchSession := newSessionForProxy(t, host, port, sessionOptions{
		disableHostLookup: true,
		queryTimeout:      15 * time.Second,
	})
	readSession := newSessionForProxy(t, host, port, sessionOptions{
		disableHostLookup: true,
		queryTimeout:      2 * time.Second,
	})

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

	table := "select_headroom_items"
	if err := batchSession.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	for _, row := range []struct {
		id   int
		name string
		idx  int
	}{
		{id: 1, name: "locked-1", idx: 1},
		{id: 2, name: "locked-2", idx: 2},
		{id: 99, name: "free", idx: 99},
	} {
		if err := batchSession.Query(insert, row.id, row.name, row.idx).Exec(); err != nil {
			t.Fatalf("seed row %d failed: %v", row.id, err)
		}
	}

	update := fmt.Sprintf("UPDATE %s.%s SET idx = sleep(3) WHERE id = ?", keyspace, table)
	batch := batchSession.NewBatch(gocql.UnloggedBatch)
	batch.Query(update, 1)
	batch.Query(update, 2)

	batchDone := make(chan error, 1)
	go func() {
		batchDone <- batchSession.ExecuteBatch(batch)
	}()

	waitFor(t, "batch workers to occupy reserved MySQL bulk slots", 5*time.Second, func() (bool, error) {
		return totalInUseConns(poolMgr) >= 2, nil
	})

	select {
	case err := <-batchDone:
		t.Fatalf("expected batch workers to remain busy during sleep window, got %v", err)
	case <-time.After(100 * time.Millisecond):
	}

	selectQuery := fmt.Sprintf("SELECT name FROM %s.%s WHERE id = ?", keyspace, table)
	start := time.Now()
	var got string
	if err := readSession.Query(selectQuery, 99).Scan(&got); err != nil {
		t.Fatalf("select on unlocked row failed: %v", err)
	}
	latency := time.Since(start)
	if got != "free" {
		t.Fatalf("unexpected select result %q", got)
	}
	if latency > time.Second {
		t.Fatalf("select should use reserved MySQL headroom, latency=%s", latency)
	}

	select {
	case err := <-batchDone:
		if err != nil {
			t.Fatalf("batch should finish after sleep window: %v", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("timed out waiting for blocked batch to finish")
	}
}

func totalInUseConns(poolMgr *mysqlbackend.PoolManager) int {
	if poolMgr == nil {
		return 0
	}
	stats := poolMgr.GetStats()
	total := 0
	for _, stat := range stats {
		if stat == nil {
			continue
		}
		total += stat.InUseConns
	}
	return total
}