//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 TestBatchRollbackOnFailure(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 := "batch_items"
	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)
	update := fmt.Sprintf("UPDATE %s.%s SET missing = ? WHERE id = ?", keyspace, table)

	batch := session.NewBatch(gocql.LoggedBatch)
	batch.Query(insert, 1, "alpha", 1)
	batch.Query(update, 2, 1)

	if err := session.ExecuteBatch(batch); err == nil {
		t.Fatalf("expected batch to fail")
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT id FROM %s.%s WHERE id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select after batch failed: %v", err)
	}
	if len(rows) != 0 {
		t.Fatalf("expected no rows after rollback")
	}
}

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

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

	batch := session.NewBatch(gocql.LoggedBatch)
	batch.SetKeyspace(keyspace)
	batch.Query(fmt.Sprintf("INSERT INTO %s (id, name, idx) VALUES (?, ?, ?)", table), 1, "alpha", 1)

	if err := session.ExecuteBatch(batch); err != nil {
		t.Fatalf("execute batch failed: %v", err)
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT id FROM %s.%s WHERE id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select after batch failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
}

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

	stmt := fmt.Sprintf(
		"BEGIN BATCH USING TIMESTAMP ? "+
			"INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?); "+
			"UPDATE %s.%s SET name = ? WHERE id = ?; "+
			"APPLY BATCH",
		keyspace,
		table,
		keyspace,
		table,
	)
	if err := session.Query(stmt, int64(1234567), 1, "alpha", 1, "beta", 1).Exec(); err != nil {
		t.Fatalf("execute batch failed: %v", err)
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT name FROM %s.%s WHERE id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select after batch failed: %v", err)
	}
	if len(rows) != 1 || asString(rows[0]["name"]) != "beta" {
		t.Fatalf("unexpected row after batch: %#v", rows)
	}
}

func TestBatchStatementMultiline(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()
	})

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

	stmt := fmt.Sprintf(
		"BEGIN BATCH\n"+
			"INSERT INTO %s.%s (id, name, idx) VALUES (1, 'alpha', 1);\n"+
			"INSERT INTO %s.%s (id, name, idx) VALUES (2, 'beta', 2);\n"+
			"APPLY BATCH",
		keyspace,
		tableA,
		keyspace,
		tableB,
	)
	if err := session.Query(stmt).Exec(); err != nil {
		t.Fatalf("execute multiline batch failed: %v", err)
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT name FROM %s.%s WHERE id = 1", keyspace, tableA))
	if err != nil {
		t.Fatalf("select from table A failed: %v", err)
	}
	if len(rows) != 1 || asString(rows[0]["name"]) != "alpha" {
		t.Fatalf("unexpected row in table A: %#v", rows)
	}

	rows, err = scanRows(session, fmt.Sprintf("SELECT name FROM %s.%s WHERE id = 2", keyspace, tableB))
	if err != nil {
		t.Fatalf("select from table B failed: %v", err)
	}
	if len(rows) != 1 || asString(rows[0]["name"]) != "beta" {
		t.Fatalf("unexpected row in table B: %#v", rows)
	}
}

func TestBatchSamePrimaryKeyInsertDifferentColumnsDataCorrectness(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 := "batch_same_pk_merge"
	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, "seed", 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)

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

	var name string
	var idx int
	var note string
	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 name != "alpha" || idx != 9 || note != "keep" {
		t.Fatalf("unexpected row after same-pk insert merge: name=%q idx=%d note=%q", name, idx, note)
	}
}

func TestCounterBatchValidation(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()
	})

	counterTable := fmt.Sprintf("CREATE TABLE %s.counter_table (id int PRIMARY KEY, cnt counter)", keyspace)
	if err := session.Query(counterTable).Exec(); err != nil {
		t.Fatalf("create counter table failed: %v", err)
	}
	standardTable := fmt.Sprintf("CREATE TABLE %s.standard_table (id int PRIMARY KEY, name text)", keyspace)
	if err := session.Query(standardTable).Exec(); err != nil {
		t.Fatalf("create standard table failed: %v", err)
	}

	loggedBatch := fmt.Sprintf(
		"BEGIN BATCH UPDATE %s.counter_table SET cnt = cnt + 1 WHERE id = 1; APPLY BATCH",
		keyspace,
	)
	requireErrorCode(t, session.Query(loggedBatch).Exec(), cql.ErrorCodeInvalid)

	counterBatch := fmt.Sprintf(
		"BEGIN COUNTER BATCH UPDATE %s.counter_table SET cnt = cnt + 1 WHERE id = 1; APPLY BATCH",
		keyspace,
	)
	if err := session.Query(counterBatch).Exec(); err != nil {
		t.Fatalf("counter batch failed: %v", err)
	}

	loggedDeleteBatch := fmt.Sprintf(
		"BEGIN BATCH DELETE FROM %s.counter_table WHERE id = 1; APPLY BATCH",
		keyspace,
	)
	requireErrorCode(t, session.Query(loggedDeleteBatch).Exec(), cql.ErrorCodeInvalid)

	counterDeleteBatch := fmt.Sprintf(
		"BEGIN COUNTER BATCH DELETE FROM %s.counter_table WHERE id = 1; APPLY BATCH",
		keyspace,
	)
	if err := session.Query(counterDeleteBatch).Exec(); err != nil {
		t.Fatalf("counter delete batch failed: %v", err)
	}

	mixedBatch := fmt.Sprintf(
		"BEGIN COUNTER BATCH UPDATE %s.counter_table SET cnt = cnt + 1 WHERE id = 1; UPDATE %s.standard_table SET name = 'x' WHERE id = 1; APPLY BATCH",
		keyspace,
		keyspace,
	)
	requireErrorCode(t, session.Query(mixedBatch).Exec(), cql.ErrorCodeInvalid)

	logged := session.NewBatch(gocql.LoggedBatch)
	logged.Query(fmt.Sprintf("UPDATE %s.counter_table SET cnt = cnt + 1 WHERE id = 1", keyspace))
	requireErrorCode(t, session.ExecuteBatch(logged), cql.ErrorCodeInvalid)

	counter := session.NewBatch(gocql.CounterBatch)
	counter.Query(fmt.Sprintf("UPDATE %s.counter_table SET cnt = cnt + 1 WHERE id = 1", keyspace))
	if err := session.ExecuteBatch(counter); err != nil {
		t.Fatalf("protocol counter batch failed: %v", err)
	}

	loggedDelete := session.NewBatch(gocql.LoggedBatch)
	loggedDelete.Query(fmt.Sprintf("DELETE FROM %s.counter_table WHERE id = 1", keyspace))
	requireErrorCode(t, session.ExecuteBatch(loggedDelete), cql.ErrorCodeInvalid)

	counterDelete := session.NewBatch(gocql.CounterBatch)
	counterDelete.Query(fmt.Sprintf("DELETE FROM %s.counter_table WHERE id = 1", keyspace))
	if err := session.ExecuteBatch(counterDelete); err != nil {
		t.Fatalf("protocol counter delete batch failed: %v", err)
	}

	mixed := session.NewBatch(gocql.CounterBatch)
	mixed.Query(fmt.Sprintf("UPDATE %s.counter_table SET cnt = cnt + 1 WHERE id = 1", keyspace))
	mixed.Query(fmt.Sprintf("UPDATE %s.standard_table SET name = 'x' WHERE id = 1", keyspace))
	requireErrorCode(t, session.ExecuteBatch(mixed), cql.ErrorCodeInvalid)
}