//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/internal/config"
)

func TestUnloggedBatchRollbackByDefault(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_unlogged_default"
	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)
	insertInvalid := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)

	batch := session.NewBatch(gocql.UnloggedBatch)
	batch.Query(insert, 1, "alpha", 1)
	// runtime failure: NULL primary key should fail during execution, not parse/prepare.
	batch.Query(insertInvalid, nil, "broken", 9)
	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 rollback for unlogged batch when parallel mode is disabled")
	}
}

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

	prevCfg := config.GetConfig()
	host, port, stop, err := startProxyWithConfig(mysqlCfg, func(cfg *config.Config) {
		cfg.Executor.UnloggedBatchParallel = true
		cfg.Executor.UnloggedBatchParallelism = 2
	})
	if err != nil {
		config.SetConfig(prevCfg)
		t.Fatalf("start proxy with unlogged parallel failed: %v", err)
	}
	defer func() {
		stop()
		config.SetConfig(prevCfg)
	}()

	session := newSessionForProxy(t, host, port, 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_unlogged_parallel"
	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)
	insertInvalid := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)

	unlogged := session.NewBatch(gocql.UnloggedBatch)
	unlogged.Query(insert, 1, "alpha", 1)
	unlogged.Query(insertInvalid, nil, "broken", 9)
	if err := session.ExecuteBatch(unlogged); err == nil {
		t.Fatalf("expected unlogged 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 unlogged batch failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected partial success for unlogged best-effort batch, got %d rows", len(rows))
	}

	logged := session.NewBatch(gocql.LoggedBatch)
	logged.Query(insert, 2, "beta", 2)
	logged.Query(insertInvalid, nil, "broken", 8)
	if err := session.ExecuteBatch(logged); err == nil {
		t.Fatalf("expected logged batch to fail")
	}
	rows, err = scanRows(session, fmt.Sprintf("SELECT id FROM %s.%s WHERE id = 2", keyspace, table))
	if err != nil {
		t.Fatalf("select after logged batch failed: %v", err)
	}
	if len(rows) != 0 {
		t.Fatalf("expected rollback for logged batch even when unlogged parallel mode is enabled")
	}
}

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

	prevCfg := config.GetConfig()
	host, port, stop, err := startProxyWithConfig(mysqlCfg, func(cfg *config.Config) {
		cfg.Executor.UnloggedBatchParallel = true
		cfg.Executor.UnloggedBatchParallelism = 2
		cfg.Executor.UnloggedMergeBeforeParallel = boolPtr(true)
		cfg.Executor.BatchInsertMergeMaxRows = 256
		cfg.Executor.BatchInsertMergeMaxBytes = 256 * 1024
	})
	if err != nil {
		config.SetConfig(prevCfg)
		t.Fatalf("start proxy with unlogged merge-before-parallel failed: %v", err)
	}
	defer func() {
		stop()
		config.SetConfig(prevCfg)
	}()

	session := newSessionForProxy(t, host, port, 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_unlogged_parallel_merge_fallback"
	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)

	unlogged := session.NewBatch(gocql.UnloggedBatch)
	unlogged.Query(insert, 1, "alpha", 1)
	unlogged.Query(insert, nil, "broken", 9)
	unlogged.Query(insert, 2, "beta", 2)
	if err := session.ExecuteBatch(unlogged); err == nil {
		t.Fatalf("expected unlogged 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 id=1 failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected id=1 to be written after merge fallback, got %d rows", len(rows))
	}
	rows, err = scanRows(session, fmt.Sprintf("SELECT id FROM %s.%s WHERE id = 2", keyspace, table))
	if err != nil {
		t.Fatalf("select id=2 failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected id=2 to be written after merge fallback, got %d rows", len(rows))
	}
}

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

	prevCfg := config.GetConfig()
	host, port, stop, err := startProxyWithConfig(mysqlCfg, func(cfg *config.Config) {
		cfg.Executor.UnloggedBatchParallel = true
		cfg.Executor.UnloggedBatchParallelism = 2
		cfg.Executor.UnloggedMergeBeforeParallel = boolPtr(false)
	})
	if err != nil {
		config.SetConfig(prevCfg)
		t.Fatalf("start proxy with unlogged merge-before-parallel disabled failed: %v", err)
	}
	defer func() {
		stop()
		config.SetConfig(prevCfg)
	}()

	session := newSessionForProxy(t, host, port, 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_unlogged_parallel_merge_off"
	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)

	unlogged := session.NewBatch(gocql.UnloggedBatch)
	unlogged.Query(insert, 1, "alpha", 1)
	unlogged.Query(insert, nil, "broken", 9)
	if err := session.ExecuteBatch(unlogged); err == nil {
		t.Fatalf("expected unlogged 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 unlogged batch failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected partial success for unlogged best-effort batch when merge is disabled, got %d rows", len(rows))
	}
}

func boolPtr(v bool) *bool {
	return &v
}