//go:build integration
// +build integration

package integration_test

import (
	"bufio"
	"fmt"
	"net"
	"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/cql"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/executor"
)

func TestPreparedCacheReuseAndInvalidation(t *testing.T) {
	requireMySQLConfig(t)

	host, port, stop, poolMgr, err := startProxyWithConfigAndPool(mysqlCfg, func(cfg *config.Config) {
		cfg.MySQL.Prepared.Enabled = true
		cfg.MySQL.Prepared.MaxStatements = 8
		cfg.MySQL.Prepared.IdleTTL = time.Minute
		cfg.MySQL.Prepared.CleanupInterval = time.Minute
		cfg.MySQL.Prepared.PrewarmOnPrepare = false
	})
	if err != nil {
		t.Fatalf("proxy start failed: %v", err)
	}
	defer stop()

	session := newSessionForHostPort(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 := "prepared_cache"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	client := newRawProtocolConnForAddr(t, host, port, cql.ProtoVersion4, rawProtocolOptions{})
	stmt := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	prepare := buildPrepareFrame(client.version, stmt, "")
	prepare.Stream = 1
	client.writeFrame(prepare)
	prepared := client.readFrame()
	stmtID := parsePreparedID(t, prepared)

	execValues := [][]byte{encodeInt32(1), encodeText("alpha"), encodeInt32(1)}
	execute := buildExecuteFrame(client.version, stmtID, nil, cql.ConsistencyOne, cql.QueryFlagValues, execValues, 0, "")
	execute.Stream = 2
	client.writeFrame(execute)
	resp := client.readFrame()
	if parseResultKind(t, resp) != executor.ResultKindVoid {
		t.Fatalf("expected VOID result")
	}

	execValues2 := [][]byte{encodeInt32(2), encodeText("beta"), encodeInt32(2)}
	execute2 := buildExecuteFrame(client.version, stmtID, nil, cql.ConsistencyOne, cql.QueryFlagValues, execValues2, 0, "")
	execute2.Stream = 3
	client.writeFrame(execute2)
	resp = client.readFrame()
	if parseResultKind(t, resp) != executor.ResultKindVoid {
		t.Fatalf("expected VOID result on second execute")
	}

	stats := poolMgr.PreparedStats()
	statKey := fmt.Sprintf("%s:%s", mysqlCfg.user, mysqlCfg.password)
	userStats, ok := stats[statKey]
	if !ok {
		t.Fatalf("missing prepared stats for user %q", statKey)
	}
	if userStats.Misses == 0 || userStats.Hits == 0 {
		t.Fatalf("expected prepared cache hit/miss, got hits=%d misses=%d", userStats.Hits, userStats.Misses)
	}

	if err := session.Query(fmt.Sprintf("ALTER TABLE %s.%s ADD extra text", keyspace, table)).Exec(); err != nil {
		t.Fatalf("alter table failed: %v", err)
	}
	stats = poolMgr.PreparedStats()
	userStats = stats[statKey]
	if userStats.Size != 0 {
		t.Fatalf("expected prepared cache cleared after DDL, size=%d", userStats.Size)
	}
}

func newSessionForHostPort(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 = 5 * 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 session failed: %v", err)
	}
	t.Cleanup(func() { session.Close() })
	return session
}

func newRawProtocolConnForAddr(t *testing.T, host string, port int, version byte, opts rawProtocolOptions) *rawProtocolConn {
	t.Helper()
	addr := fmt.Sprintf("%s:%d", host, port)
	conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
	if err != nil {
		t.Fatalf("dial proxy failed: %v", err)
	}
	client := &rawProtocolConn{
		t:       t,
		conn:    conn,
		version: version,
		reader:  bufio.NewReader(conn),
	}
	t.Cleanup(func() {
		_ = conn.Close()
	})
	client.startup(opts)
	return client
}