//go:build integration
// +build integration

package integration_test

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

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

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

	session := newSession(t, sessionOptions{disableHostLookup: true})
	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()
	})

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

	t.Run("hash-single-int", func(t *testing.T) {
		table := "hash_single_int"
		create := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY, name text)", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		ddl := fetchCreateTable(t, db, keyspace, table)
		if !strings.Contains(ddl, "PARTITION BY HASH") {
			t.Fatalf("expected HASH partition clause, got: %s", ddl)
		}
		expected := expectedPartitionSize()
		if !strings.Contains(ddl, fmt.Sprintf("PARTITIONS %d", expected)) {
			t.Fatalf("expected partition count %d, got: %s", expected, ddl)
		}
	})

	t.Run("key-single-varint", func(t *testing.T) {
		table := "key_single_varint"
		create := fmt.Sprintf("CREATE TABLE %s.%s (id varint PRIMARY KEY, name text)", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		ddl := fetchCreateTable(t, db, keyspace, table)
		if !strings.Contains(ddl, "PARTITION BY KEY") {
			t.Fatalf("expected KEY partition clause, got: %s", ddl)
		}
		expected := expectedPartitionSize()
		if !strings.Contains(ddl, fmt.Sprintf("PARTITIONS %d", expected)) {
			t.Fatalf("expected partition count %d, got: %s", expected, ddl)
		}
	})

	t.Run("key-composite-partition-key", func(t *testing.T) {
		table := "key_composite"
		create := fmt.Sprintf("CREATE TABLE %s.%s (a int, b int, c int, PRIMARY KEY ((a, b), c))", keyspace, table)
		if err := session.Query(create).Exec(); err != nil {
			t.Fatalf("create table failed: %v", err)
		}
		ddl := fetchCreateTable(t, db, keyspace, table)
		if !strings.Contains(ddl, "PARTITION BY KEY") {
			t.Fatalf("expected KEY partition clause, got: %s", ddl)
		}
		expected := expectedPartitionSize()
		if !strings.Contains(ddl, fmt.Sprintf("PARTITIONS %d", expected)) {
			t.Fatalf("expected partition count %d, got: %s", expected, ddl)
		}
	})
}

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

	session := newSession(t, sessionOptions{disableHostLookup: true})
	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()
	})
	db, err := sql.Open("mysql", mysqlDSN(mysqlCfg))
	if err != nil {
		t.Fatalf("mysql connect failed: %v", err)
	}
	defer db.Close()

	table := "text_partition_key"
	create := fmt.Sprintf("CREATE TABLE %s.%s (id text PRIMARY KEY)", keyspace, table)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	ddl := fetchCreateTable(t, db, keyspace, table)
	if strings.Contains(ddl, "PARTITION BY") {
		t.Fatalf("expected no PARTITION BY for text partition key, got: %s", ddl)
	}
}

func fetchCreateTable(t *testing.T, db *sql.DB, schema, table string) string {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	query := fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", schema, table)
	var tableName string
	var ddl string
	if err := db.QueryRowContext(ctx, query).Scan(&tableName, &ddl); err != nil {
		t.Fatalf("show create table failed: %v", err)
	}
	return ddl
}

func expectedPartitionSize() int {
	cfg := config.GetConfig()
	if cfg == nil {
		return 10
	}
	if cfg.MySQL.PartitionDefaultSize <= 0 {
		return 10
	}
	return cfg.MySQL.PartitionDefaultSize
}