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

func newSessionForProxy(t *testing.T, host string, port int, opts sessionOptions) *gocql.Session {
	t.Helper()
	requireProxy(t)

	if opts.protoVersion == 0 {
		opts.protoVersion = 4
	}
	if opts.numConns == 0 {
		opts.numConns = 1
	}
	if opts.connectTimeout == 0 {
		opts.connectTimeout = 5 * time.Second
	}
	if opts.queryTimeout == 0 {
		opts.queryTimeout = 5 * time.Second
	}

	cluster := gocql.NewCluster(host)
	cluster.Port = port
	cluster.ProtoVersion = opts.protoVersion
	cluster.NumConns = opts.numConns
	cluster.ConnectTimeout = opts.connectTimeout
	cluster.Timeout = opts.queryTimeout
	cluster.Keyspace = opts.keyspace
	cluster.DisableInitialHostLookup = opts.disableHostLookup
	if opts.compressor != nil {
		cluster.Compressor = opts.compressor
	}

	if !opts.skipAuth {
		username := opts.username
		if username == "" {
			username = mysqlCfg.user
		}
		password := opts.password
		if password == "" {
			password = mysqlCfg.password
		}

		allowed := opts.allowedAuthClasses
		if len(allowed) == 0 {
			allowed = []string{"org.apache.cassandra.auth.PasswordAuthenticator"}
		}
		cluster.Authenticator = gocql.PasswordAuthenticator{
			Username:              username,
			Password:              password,
			AllowedAuthenticators: allowed,
		}
	}

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

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

	prevCfg := config.GetConfig()
	hostB, portB, stopB, err := startProxyWithConfig(mysqlCfg, func(cfg *config.Config) {
		cfg.SchemaSync.CheckInterval = 200 * time.Millisecond
	})
	if err != nil {
		config.SetConfig(prevCfg)
		t.Fatalf("start second proxy failed: %v", err)
	}
	defer func() {
		stopB()
		config.SetConfig(prevCfg)
	}()

	sessionA := newSession(t, sessionOptions{})
	sessionB := newSessionForProxy(t, hostB, portB, sessionOptions{})

	keyspace := uniqueKeyspace()
	table := "schema_sync_tbl"

	if _, err := scanRows(sessionB, "SELECT table_name FROM system_schema.tables"); err != nil {
		t.Fatalf("warm schema cache failed: %v", err)
	}

	if err := sessionA.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		t.Fatalf("create keyspace failed: %v", err)
	}
	defer sessionA.Query(dropKeyspaceCQL(keyspace)).Exec()

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

	waitFor(t, "schema sync table visibility", 3*time.Second, func() (bool, error) {
		query := fmt.Sprintf(
			"SELECT table_name FROM system_schema.tables WHERE keyspace_name = '%s' AND table_name = '%s'",
			keyspace,
			table,
		)
		iter := sessionB.Query(query).Iter()
		row := map[string]interface{}{}
		ok := iter.MapScan(row)
		if err := iter.Close(); err != nil {
			return false, err
		}
		return ok, nil
	})
}

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

	prevCfg := config.GetConfig()
	hostB, portB, stopB, err := startProxyWithConfig(mysqlCfg, func(cfg *config.Config) {
		cfg.SchemaSync.CheckInterval = 200 * time.Millisecond
	})
	if err != nil {
		config.SetConfig(prevCfg)
		t.Fatalf("start second proxy failed: %v", err)
	}
	defer func() {
		stopB()
		config.SetConfig(prevCfg)
	}()

	sessionA := newSession(t, sessionOptions{})
	sessionB := newSessionForProxy(t, hostB, portB, sessionOptions{})

	if _, err := scanRows(sessionB, "SELECT schema_version FROM system.local"); err != nil {
		t.Fatalf("warm schema_version cache failed: %v", err)
	}

	keyspace := uniqueKeyspace()
	table := "schema_sync_page"

	if err := sessionA.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		t.Fatalf("create keyspace failed: %v", err)
	}
	defer sessionA.Query(dropKeyspaceCQL(keyspace)).Exec()

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

	for i := 0; i < 5; i++ {
		if err := sessionA.Query(
			fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table),
			i,
			fmt.Sprintf("name_%d", i),
			i,
		).Exec(); err != nil {
			t.Fatalf("insert row failed: %v", err)
		}
	}

	iter := sessionA.Query(
		fmt.Sprintf("SELECT id, name FROM %s.%s", keyspace, table),
	).PageSize(2).Iter()
	var id int
	var name string
	read := 0
	for iter.Scan(&id, &name) {
		read++
		if read >= 2 {
			break
		}
	}
	pageState := iter.PageState()
	if err := iter.Close(); err != nil {
		t.Fatalf("paged query failed: %v", err)
	}
	if len(pageState) == 0 {
		t.Fatalf("expected paging state from first page")
	}

	time.Sleep(300 * time.Millisecond)

	iterB := sessionB.Query(
		fmt.Sprintf("SELECT id, name FROM %s.%s", keyspace, table),
	).PageSize(2).PageState(pageState).Iter()
	for iterB.Scan(&id, &name) {
	}
	if err := iterB.Close(); err != nil {
		t.Fatalf("paged query on second proxy failed: %v", err)
	}
}