//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"

	"github.com/apache/cassandra-gocql-driver/v2"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql"
)

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}
	if err := session.Query(insert, 1, gocql.UnsetValue, 2).Exec(); err != nil {
		t.Fatalf("insert with unset failed: %v", err)
	}

	var name []byte
	var idx int
	selectStmt := fmt.Sprintf("SELECT name, idx FROM %s.%s WHERE id = 1", keyspace, table)
	if err := session.Query(selectStmt).Scan(&name, &idx); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if string(name) != "alpha" {
		t.Fatalf("expected name to remain 'alpha', got %v", name)
	}
	if idx != 2 {
		t.Fatalf("unexpected idx after insert: %v", idx)
	}
}

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	update := fmt.Sprintf("UPDATE %s.%s SET name = ?, idx = ? WHERE id = ?", keyspace, table)
	if err := session.Query(update, gocql.UnsetValue, gocql.UnsetValue, 1).Exec(); err != nil {
		t.Fatalf("update failed: %v", err)
	}

	var name []byte
	var idx int
	selectStmt := fmt.Sprintf("SELECT name, idx FROM %s.%s WHERE id = 1", keyspace, table)
	if err := session.Query(selectStmt).Scan(&name, &idx); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if string(name) != "alpha" {
		t.Fatalf("expected name to remain 'alpha', got %v", name)
	}
	if idx != 1 {
		t.Fatalf("unexpected idx after update: %v", idx)
	}
}

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	update := fmt.Sprintf("UPDATE %s.%s SET name = ?, idx = ? WHERE id = ?", keyspace, table)
	if err := session.Query(update, nil, gocql.UnsetValue, 1).Exec(); err != nil {
		t.Fatalf("update failed: %v", err)
	}

	var name []byte
	var idx int
	selectStmt := fmt.Sprintf("SELECT name, idx FROM %s.%s WHERE id = 1", keyspace, table)
	if err := session.Query(selectStmt).Scan(&name, &idx); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if name != nil {
		t.Fatalf("expected name to be null, got %v", name)
	}
	if idx != 1 {
		t.Fatalf("unexpected idx after update: %v", idx)
	}
}

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	update := fmt.Sprintf("UPDATE %s.%s SET name = 'beta' WHERE id = ?", keyspace, table)
	err := session.Query(update, gocql.UnsetValue).Exec()
	requireErrorCode(t, err, cql.ErrorCodeInvalid)

	updateIf := fmt.Sprintf("UPDATE %s.%s SET name = 'beta' WHERE id = 1 IF name = ?", keyspace, table)
	err = session.Query(updateIf, gocql.UnsetValue).Exec()
	requireErrorCode(t, err, cql.ErrorCodeInvalid)

	insertPK := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	err = session.Query(insertPK, gocql.UnsetValue, "gamma", 2).Exec()
	requireErrorCode(t, err, cql.ErrorCodeInvalid)
}

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	batch := session.NewBatch(gocql.LoggedBatch)
	update := fmt.Sprintf("UPDATE %s.%s SET name = ?, idx = ? WHERE id = ?", keyspace, table)
	batch.Query(update, gocql.UnsetValue, 2, 1)
	if err := session.ExecuteBatch(batch); err != nil {
		t.Fatalf("batch update failed: %v", err)
	}

	var name []byte
	var idx int
	selectStmt := fmt.Sprintf("SELECT name, idx FROM %s.%s WHERE id = 1", keyspace, table)
	if err := session.Query(selectStmt).Scan(&name, &idx); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if string(name) != "alpha" {
		t.Fatalf("expected name to remain 'alpha', got %v", name)
	}
	if idx != 2 {
		t.Fatalf("unexpected idx after batch: %v", idx)
	}
}

func TestUnsetBatchSamePrimaryKeyInsertTemplatesDataCorrectness(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 := "unset_batch_same_pk_insert"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (id int PRIMARY KEY, name text, idx int, note text)",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	seed := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx, note) VALUES (?, ?, ?, ?)", keyspace, table)
	if err := session.Query(seed, 1, "alpha", 1, "keep").Exec(); err != nil {
		t.Fatalf("seed row failed: %v", err)
	}

	insertName := fmt.Sprintf("INSERT INTO %s.%s (id, name) VALUES (?, ?)", keyspace, table)
	insertIdx := fmt.Sprintf("INSERT INTO %s.%s (id, idx) VALUES (?, ?)", keyspace, table)
	insertNote := fmt.Sprintf("INSERT INTO %s.%s (id, note) VALUES (?, ?)", keyspace, table)

	batch := session.NewBatch(gocql.LoggedBatch)
	batch.Query(insertName, 1, gocql.UnsetValue)
	batch.Query(insertIdx, 1, 2)
	batch.Query(insertNote, 1, nil)
	if err := session.ExecuteBatch(batch); err != nil {
		t.Fatalf("execute batch failed: %v", err)
	}

	var name []byte
	var idx int
	var note []byte
	selectSQL := fmt.Sprintf("SELECT name, idx, note FROM %s.%s WHERE id = ?", keyspace, table)
	if err := session.Query(selectSQL, 1).Scan(&name, &idx, &note); err != nil {
		t.Fatalf("select after batch failed: %v", err)
	}
	if string(name) != "alpha" {
		t.Fatalf("expected name to remain alpha, got %v", name)
	}
	if idx != 2 {
		t.Fatalf("expected idx to become 2, got %d", idx)
	}
	if note != nil {
		t.Fatalf("expected note to become null, got %v", note)
	}
}