//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"strconv"
	"sync"
	"testing"
)

func TestConcurrentQueries(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 := "concurrent_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)
	selectStmt := fmt.Sprintf("SELECT name FROM %s.%s WHERE id = ?", keyspace, table)

	const workers = 8
	const perWorker = 5

	var wg sync.WaitGroup
	errCh := make(chan error, workers*perWorker)

	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func(worker int) {
			defer wg.Done()
			for i := 0; i < perWorker; i++ {
				id := worker*perWorker + i
				name := "name-" + strconv.Itoa(id)
				if err := session.Query(insert, id, name, id).Exec(); err != nil {
					errCh <- err
					return
				}
				var got string
				if err := session.Query(selectStmt, id).Scan(&got); err != nil {
					errCh <- err
					return
				}
				if got != name {
					errCh <- fmt.Errorf("unexpected name: %q (want %q)", got, name)
					return
				}
			}
		}(w)
	}

	wg.Wait()
	close(errCh)
	for err := range errCh {
		if err != nil {
			t.Fatalf("concurrent query failed: %v", err)
		}
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT COUNT(*) AS cnt FROM %s.%s", keyspace, table))
	if err != nil {
		t.Fatalf("count query failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected 1 row for count, got %d", len(rows))
	}
	if got := asString(rows[0]["cnt"]); got != strconv.Itoa(workers*perWorker) {
		t.Fatalf("unexpected count: %s", got)
	}
}