//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"

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

func TestCollectionMutationSemantics(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 := "collection_items"
	create := fmt.Sprintf(
		"CREATE TABLE %s.%s (id int PRIMARY KEY, name text, tags list<text>, roles set<text>, attrs map<text, text>)",
		keyspace,
		table,
	)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, tags, roles, attrs) VALUES (?, ?, ?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", []string{"a"}, []string{"admin"}, map[string]string{"email": "a"}).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	update := fmt.Sprintf("UPDATE %s.%s SET tags = tags + ['b'] WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("list append failed: %v", err)
	}
	row := mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringSlice(t, row.tags, []string{"a", "b"})

	update = fmt.Sprintf("UPDATE %s.%s SET tags = ['z'] + tags WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("list prepend failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringSlice(t, row.tags, []string{"z", "a", "b"})

	update = fmt.Sprintf("UPDATE %s.%s SET tags[1] = 'x' WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("list index update failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringSlice(t, row.tags, []string{"z", "x", "b"})

	update = fmt.Sprintf("UPDATE %s.%s SET roles = roles + {'user', 'admin'} WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("set add failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringSet(t, row.roles, []string{"admin", "user"})

	update = fmt.Sprintf("UPDATE %s.%s SET roles = roles - {'admin'} WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("set remove failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringSet(t, row.roles, []string{"user"})

	update = fmt.Sprintf("UPDATE %s.%s SET attrs = attrs + {'name': 'bob'} WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("map merge failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringMap(t, row.attrs, map[string]string{"email": "a", "name": "bob"})

	update = fmt.Sprintf("UPDATE %s.%s SET attrs['email'] = 'b' WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("map key update failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringMap(t, row.attrs, map[string]string{"email": "b", "name": "bob"})

	update = fmt.Sprintf("UPDATE %s.%s SET attrs = attrs - {'name'} WHERE id = 1", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("map remove failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringMap(t, row.attrs, map[string]string{"email": "b"})

	remove := fmt.Sprintf("DELETE attrs['email'] FROM %s.%s WHERE id = 1", keyspace, table)
	if err := session.Query(remove).Exec(); err != nil {
		t.Fatalf("map key delete failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 1)
	requireStringMap(t, row.attrs, map[string]string{})

	update = fmt.Sprintf("UPDATE %s.%s SET name = 'new', tags = tags + ['first'], roles = roles + {'member'}, attrs['k'] = 'v' WHERE id = 99", keyspace, table)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("update missing row failed: %v", err)
	}
	row = mustFetchCollectionRow(t, session, keyspace, table, 99)
	if row.name != "new" {
		t.Fatalf("unexpected name on upsert: %v", row.name)
	}
	requireStringSlice(t, row.tags, []string{"first"})
	requireStringSet(t, row.roles, []string{"member"})
	requireStringMap(t, row.attrs, map[string]string{"k": "v"})
}

func TestCollectionLiteralInsert(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 := "collection_literals"
	create := fmt.Sprintf(
		"CREATE TABLE %s.%s (id int PRIMARY KEY, name text, tags list<text>, roles set<text>, attrs map<text, text>)",
		keyspace,
		table,
	)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	insert := fmt.Sprintf(
		"INSERT INTO %s.%s (id, name, tags, roles, attrs) VALUES (1, 'alpha', ['x', 'y'], {'admin', 'user'}, {'region': 'HK', 'lang': 'zh'})",
		keyspace,
		table,
	)
	if err := session.Query(insert).Exec(); err != nil {
		t.Fatalf("literal insert failed: %v", err)
	}

	row := mustFetchCollectionRow(t, session, keyspace, table, 1)
	if row.name != "alpha" {
		t.Fatalf("unexpected name: %v", row.name)
	}
	requireStringSlice(t, row.tags, []string{"x", "y"})
	requireStringSet(t, row.roles, []string{"admin", "user"})
	requireStringMap(t, row.attrs, map[string]string{"region": "HK", "lang": "zh"})
}

type collectionRow struct {
	name  string
	tags  []string
	roles []string
	attrs map[string]string
}

func mustFetchCollectionRow(t *testing.T, session *gocql.Session, keyspace, table string, id int) collectionRow {
	t.Helper()
	query := fmt.Sprintf("SELECT name, tags, roles, attrs FROM %s.%s WHERE id = %d", keyspace, table, id)
	rows, err := scanRows(session, query)
	if err != nil {
		t.Fatalf("select row failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	row := rows[0]
	return collectionRow{
		name:  asString(row["name"]),
		tags:  stringList(row["tags"]),
		roles: stringList(row["roles"]),
		attrs: stringMap(row["attrs"]),
	}
}

func requireStringSlice(t *testing.T, got, expected []string) {
	t.Helper()
	if len(got) != len(expected) {
		t.Fatalf("expected %d items, got %d", len(expected), len(got))
	}
	for i := range expected {
		if got[i] != expected[i] {
			t.Fatalf("expected %v, got %v", expected, got)
		}
	}
}

func requireStringSet(t *testing.T, got, expected []string) {
	t.Helper()
	gotSet := make(map[string]struct{}, len(got))
	for _, value := range got {
		gotSet[value] = struct{}{}
	}
	for _, value := range expected {
		if _, ok := gotSet[value]; !ok {
			t.Fatalf("expected set to contain %q, got %v", value, got)
		}
		delete(gotSet, value)
	}
	if len(gotSet) != 0 {
		left := make([]string, 0, len(gotSet))
		for key := range gotSet {
			left = append(left, key)
		}
		t.Fatalf("unexpected set values: %v", left)
	}
}

func requireStringMap(t *testing.T, got, expected map[string]string) {
	t.Helper()
	if len(got) != len(expected) {
		t.Fatalf("expected %d entries, got %d", len(expected), len(got))
	}
	for key, value := range expected {
		if got[key] != value {
			t.Fatalf("expected %s=%s, got %v", key, value, got)
		}
	}
}

func stringMap(value interface{}) map[string]string {
	switch v := value.(type) {
	case map[string]string:
		return v
	case map[string]interface{}:
		out := make(map[string]string, len(v))
		for key, val := range v {
			out[key] = asString(val)
		}
		return out
	case map[interface{}]interface{}:
		out := make(map[string]string, len(v))
		for key, val := range v {
			out[asString(key)] = asString(val)
		}
		return out
	default:
		return nil
	}
}