//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"
	"time"

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

func TestDDLAlreadyExists(t *testing.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()
	})

	if err := session.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		requireErrorCode(t, err, cql.ErrorCodeAlreadyExists)
	} else {
		t.Fatalf("expected create keyspace duplicate to fail")
	}

	table := "items"
	createTable := createTableCQL(keyspace, table)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	if err := session.Query(createTable).Exec(); err != nil {
		requireErrorCode(t, err, cql.ErrorCodeAlreadyExists)
	} else {
		t.Fatalf("expected create table duplicate to fail")
	}

	index := "items_idx_idx"
	createIndex := createIndexCQL(index, keyspace, table, "idx")
	if err := session.Query(createIndex).Exec(); err != nil {
		t.Fatalf("create index failed: %v", err)
	}
	if err := session.Query(createIndex).Exec(); err != nil {
		requireErrorCode(t, err, cql.ErrorCodeAlreadyExists)
	} else {
		t.Fatalf("expected create index duplicate to fail")
	}
}

func TestDDLIfExistsSemantics(t *testing.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()
	})

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

	if err := session.Query(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY)", keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table if not exists failed: %v", err)
	}
	if err := session.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s.%s", keyspace, table)).Exec(); err != nil {
		t.Fatalf("drop table if exists failed: %v", err)
	}
	if err := session.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s.%s", keyspace, table)).Exec(); err != nil {
		t.Fatalf("drop table if exists (missing) failed: %v", err)
	}
}

func TestReservedKeyspacesInvalid(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	reserved := []string{
		"system",
		"system_schema",
		"mysql",
		"performance_schema",
		"sys",
		"metrics_schema",
	}

	for _, name := range reserved {
		t.Run(name, func(t *testing.T) {
			cqlText := createKeyspaceCQL(name)
			err := session.Query(cqlText).Exec()
			requireErrorCode(t, err, cql.ErrorCodeInvalid)
		})
	}
}

func TestInformationSchemaAccessible(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	rows, err := scanRows(session, "SELECT table_name FROM information_schema.tables LIMIT 1 ALLOW FILTERING")
	if err != nil {
		t.Fatalf("query information_schema failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatal("expected information_schema.tables row")
	}
}

func TestDDLQueryUsesQueryOpcode(t *testing.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()
	})

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

	waitFor(t, "ddl table visible", 5*time.Second, func() (bool, error) {
		iter := session.Query(fmt.Sprintf("SELECT table_name FROM system_schema.tables WHERE keyspace_name = '%s'", keyspace)).Iter()
		row := map[string]interface{}{}
		for iter.MapScan(row) {
			if asString(row["table_name"]) == table {
				_ = iter.Close()
				return true, nil
			}
			row = map[string]interface{}{}
		}
		return false, iter.Close()
	})
}

func TestErrorCodeInvalidOnUnsupportedDDL(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	err := session.Query("CREATE INDEX IF NOT EXISTS idx ON system.local (key)").Exec()
	requireErrorCode(t, err, cql.ErrorCodeInvalid)
}