//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"

	gocql "github.com/apache/cassandra-gocql-driver/v2"

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

func TestIndexSelectionWithoutPredicateAllowed(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 := "full_scan"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, PRIMARY KEY (a))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	query := fmt.Sprintf("SELECT * FROM %s.%s", keyspace, table)
	if err := runSelect(session, query); err != nil {
		t.Fatalf("expected query without predicate to succeed: %v", err)
	}
}

func TestIndexSelectionEqualityColumnsMustBeInIndex(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 := "eq_match"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, c int, d int, PRIMARY KEY (a, b, c))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	query := fmt.Sprintf("SELECT * FROM %s.%s WHERE a = 1 AND d = 2", keyspace, table)
	err := runSelect(session, query)
	requireErrorCode(t, err, cql.ErrorCodeInvalid)

	okQuery := fmt.Sprintf("SELECT * FROM %s.%s WHERE a = 1 AND b = 2", keyspace, table)
	if err := runSelect(session, okQuery); err != nil {
		t.Fatalf("expected query to succeed with primary index: %v", err)
	}
}

func TestIndexSelectionRangeRequirement(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 := "range_group"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, c int, d int, PRIMARY KEY (a, b, c))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	rangeQuery := fmt.Sprintf("SELECT * FROM %s.%s WHERE a = 1 AND d > 5", keyspace, table)
	err := runSelect(session, rangeQuery)
	requireErrorCode(t, err, cql.ErrorCodeInvalid)

	okRangeQuery := fmt.Sprintf("SELECT * FROM %s.%s WHERE a = 1 AND b > 2", keyspace, table)
	if err := runSelect(session, okRangeQuery); err != nil {
		t.Fatalf("expected range query to succeed on primary index: %v", err)
	}

}

func TestIndexSelectionGroupByAllowed(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 := "group_by"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, c int, PRIMARY KEY (a, b, c))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	groupByQuery := fmt.Sprintf("SELECT c, count(*) FROM %s.%s WHERE a = 1 AND b = 2 GROUP BY c", keyspace, table)
	if err := runSelect(session, groupByQuery); err != nil {
		t.Fatalf("expected group by query to succeed: %v", err)
	}
}

func TestIndexSelectionSecondaryIndexEquality(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 := "sec_eq"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, c int, PRIMARY KEY (a))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	if err := session.Query(createIndexCQL("idx_b", keyspace, table, "b")).Exec(); err != nil {
		t.Fatalf("create secondary index failed: %v", err)
	}

	query := fmt.Sprintf("SELECT * FROM %s.%s WHERE b = 1", keyspace, table)
	if err := runSelect(session, query); err != nil {
		t.Fatalf("expected secondary index equality query to succeed: %v", err)
	}
}

func TestIndexSelectionSecondaryIndexRange(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 := "sec_range"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, c int, PRIMARY KEY (a))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	if err := session.Query(createIndexCQL("idx_b", keyspace, table, "b")).Exec(); err != nil {
		t.Fatalf("create secondary index failed: %v", err)
	}

	query := fmt.Sprintf("SELECT * FROM %s.%s WHERE b > 1", keyspace, table)
	if err := runSelect(session, query); err != nil {
		t.Fatalf("expected secondary index range query to succeed: %v", err)
	}
}

func TestIndexSelectionSecondaryIndexFailsWithExtraEquality(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 := "sec_eq_fail"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (a int, b int, c int, PRIMARY KEY (a))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	if err := session.Query(createIndexCQL("idx_b", keyspace, table, "b")).Exec(); err != nil {
		t.Fatalf("create secondary index failed: %v", err)
	}

	query := fmt.Sprintf("SELECT * FROM %s.%s WHERE b = 1 AND c = 2", keyspace, table)
	err := runSelect(session, query)
	requireErrorCode(t, err, cql.ErrorCodeInvalid)
}

func runSelect(session *gocql.Session, query string) error {
	iter := session.Query(query).Iter()
	return iter.Close()
}