//go:build integration
// +build integration

package integration_test

import (
	"encoding/json"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"testing"
)

func TestDMLUpsertOverwrite(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 := "upsert_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)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert row1 failed: %v", err)
	}
	if err := session.Query(insert, 1, "beta", 2).Exec(); err != nil {
		t.Fatalf("insert row1 overwrite failed: %v", err)
	}

	var name string
	var idx int
	selectStmt := fmt.Sprintf("SELECT name, idx FROM %s.%s WHERE id = ?", keyspace, table)
	if err := session.Query(selectStmt, 1).Scan(&name, &idx); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if name != "beta" || idx != 2 {
		t.Fatalf("unexpected values: name=%q idx=%d", name, idx)
	}
}

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

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

	rows, err := scanRows(session, fmt.Sprintf("SELECT JSON id, name, idx FROM %s.%s WHERE id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select JSON failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	payload := asString(rows[0]["json"])
	var decoded map[string]interface{}
	if err := json.Unmarshal([]byte(payload), &decoded); err != nil {
		t.Fatalf("decode JSON failed: %v", err)
	}
	if decoded["name"] != "alpha" || int(decoded["id"].(float64)) != 1 || int(decoded["idx"].(float64)) != 7 {
		t.Fatalf("unexpected JSON payload: %#v", decoded)
	}

	rows, err = scanRows(session, fmt.Sprintf("SELECT to_json(name) AS name_json FROM %s.%s WHERE id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select to_json failed: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	nameJSON := asString(rows[0]["name_json"])
	var name string
	if err := json.Unmarshal([]byte(nameJSON), &name); err != nil {
		t.Fatalf("decode to_json failed: %v", err)
	}
	if name != "alpha" {
		t.Fatalf("unexpected to_json value: %q", name)
	}
}

func TestDMLGroupBy(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 := "group_items"
	create := fmt.Sprintf("CREATE TABLE %s.%s (id int, bucket int, value int, PRIMARY KEY (id, bucket))", 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, bucket, value) VALUES (?, ?, ?)", keyspace, table)
	for bucket := 1; bucket <= 3; bucket++ {
		if err := session.Query(insert, 1, bucket, bucket).Exec(); err != nil {
			t.Fatalf("insert id=1 bucket=%d failed: %v", bucket, err)
		}
	}
	for bucket := 1; bucket <= 2; bucket++ {
		if err := session.Query(insert, 2, bucket, bucket).Exec(); err != nil {
			t.Fatalf("insert id=2 bucket=%d failed: %v", bucket, err)
		}
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT id, COUNT(*) AS cnt FROM %s.%s GROUP BY id", keyspace, table))
	if err != nil {
		t.Fatalf("group by query failed: %v", err)
	}
	if len(rows) != 2 {
		t.Fatalf("expected 2 rows, got %d", len(rows))
	}
	counts := map[int]int{}
	for _, row := range rows {
		id, _ := strconv.Atoi(asString(row["id"]))
		cnt, _ := strconv.Atoi(asString(row["cnt"]))
		counts[id] = cnt
	}
	if counts[1] != 3 || counts[2] != 2 {
		t.Fatalf("unexpected group counts: %#v", counts)
	}
}

func TestDMLPerPartitionLimit(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 := "partition_limit"
	create := fmt.Sprintf("CREATE TABLE %s.%s (id int, bucket int, value int, PRIMARY KEY (id, bucket))", 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, bucket, value) VALUES (?, ?, ?)", keyspace, table)
	for id := 1; id <= 3; id++ {
		for bucket := 1; bucket <= 2; bucket++ {
			if err := session.Query(insert, id, bucket, id*10+bucket).Exec(); err != nil {
				t.Fatalf("insert failed: %v", err)
			}
		}
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT id, bucket FROM %s.%s ORDER BY id, bucket PER PARTITION LIMIT 2", keyspace, table))
	if err != nil {
		t.Fatalf("per partition limit query failed: %v", err)
	}
	if len(rows) != 2 {
		t.Fatalf("expected 2 rows, got %d", len(rows))
	}
}

func TestDMLColumnDelete(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 := "column_delete"
	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)
	}

	deleteStmt := fmt.Sprintf("DELETE name FROM %s.%s WHERE id = ?", keyspace, table)
	if err := session.Query(deleteStmt, 1).Exec(); err != nil {
		t.Fatalf("column delete failed: %v", err)
	}

	var name []byte
	var idx int
	if err := session.Query(fmt.Sprintf("SELECT name, idx FROM %s.%s WHERE id = 1", keyspace, table)).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 value: %v", idx)
	}
}

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

	insertIf := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (1, 'alpha', 1) IF NOT EXISTS", keyspace, table)
	cas := map[string]interface{}{}
	applied, err := session.Query(insertIf).MapScanCAS(cas)
	if err != nil {
		t.Fatalf("insert IF NOT EXISTS failed: %v", err)
	}
	if !applied {
		t.Fatalf("expected applied on first insert")
	}

	cas = map[string]interface{}{}
	applied, err = session.Query(insertIf).MapScanCAS(cas)
	if err != nil {
		t.Fatalf("insert IF NOT EXISTS failed: %v", err)
	}
	if applied {
		t.Fatalf("expected not applied on duplicate insert")
	}

	updateIf := fmt.Sprintf("UPDATE %s.%s SET name = 'beta' WHERE id = 1 IF EXISTS", keyspace, table)
	cas = map[string]interface{}{}
	applied, err = session.Query(updateIf).MapScanCAS(cas)
	if err != nil {
		t.Fatalf("update IF EXISTS failed: %v", err)
	}
	if !applied {
		t.Fatalf("expected applied on update IF EXISTS")
	}

	updateCond := fmt.Sprintf("UPDATE %s.%s SET name = 'gamma' WHERE id = 1 IF name = 'beta'", keyspace, table)
	cas = map[string]interface{}{}
	applied, err = session.Query(updateCond).MapScanCAS(cas)
	if err != nil {
		t.Fatalf("conditional update failed: %v", err)
	}
	if !applied {
		t.Fatalf("expected applied for matching IF condition")
	}

	updateCond = fmt.Sprintf("UPDATE %s.%s SET name = 'delta' WHERE id = 1 IF name = 'missing'", keyspace, table)
	cas = map[string]interface{}{}
	applied, err = session.Query(updateCond).MapScanCAS(cas)
	if err != nil {
		t.Fatalf("conditional update failed: %v", err)
	}
	if applied {
		t.Fatalf("expected not applied for mismatched IF condition")
	}

	updateMissing := fmt.Sprintf("UPDATE %s.%s SET name = 'noop' WHERE id = 2 IF EXISTS", keyspace, table)
	cas = map[string]interface{}{}
	applied, err = session.Query(updateMissing).MapScanCAS(cas)
	if err != nil {
		t.Fatalf("update IF EXISTS missing row failed: %v", err)
	}
	if applied {
		t.Fatalf("expected not applied for missing row")
	}
}

func TestDMLAllowFilteringAndToken(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 := "filter_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)
	for i := 1; i <= 3; i++ {
		if err := session.Query(insert, i, "alpha", i).Exec(); err != nil {
			t.Fatalf("insert failed: %v", err)
		}
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT id FROM %s.%s WHERE name = 'alpha' ALLOW FILTERING", keyspace, table))
	if err != nil {
		t.Fatalf("allow filtering query failed: %v", err)
	}
	if len(rows) != 3 {
		t.Fatalf("expected 3 rows, got %d", len(rows))
	}

	rows, err = scanRows(session, fmt.Sprintf("SELECT id FROM %s.%s WHERE token(id) > 1 ORDER BY id ALLOW FILTERING", keyspace, table))
	if err != nil {
		t.Fatalf("token query failed: %v", err)
	}
	got := make([]int, 0, len(rows))
	for _, row := range rows {
		id, _ := strconv.Atoi(asString(row["id"]))
		got = append(got, id)
	}
	sort.Ints(got)
	expected := []int{2, 3}
	if len(got) != len(expected) {
		t.Fatalf("unexpected token results: %#v", got)
	}
	for i := range expected {
		if got[i] != expected[i] {
			t.Fatalf("unexpected token results: %#v", got)
		}
	}
}

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

	batch := fmt.Sprintf("BEGIN BATCH INSERT INTO %s.%s (id, name, idx) VALUES (1, 'alpha', 1); UPDATE %s.%s SET missing = 1 WHERE id = 1; APPLY BATCH", keyspace, table, keyspace, table)
	if err := session.Query(batch).Exec(); err == nil {
		t.Fatalf("expected batch statement 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 from batch statement")
	}
}

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

	create := fmt.Sprintf("CREATE TABLE %s.case_items (\"Id\" int PRIMARY KEY, \"CamelCase\" text)", keyspace)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

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

	var value string
	selectStmt := fmt.Sprintf("SELECT \"CamelCase\" FROM %s.case_items WHERE \"Id\" = ?", keyspace)
	if err := session.Query(selectStmt, 1).Scan(&value); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if value != "alpha" {
		t.Fatalf("unexpected value: %q", value)
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT column_name FROM system_schema.columns WHERE keyspace_name = '%s' AND table_name = 'case_items'", keyspace))
	if err != nil {
		t.Fatalf("system_schema columns query failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected system_schema columns rows")
	}
	found := map[string]bool{}
	for _, row := range rows {
		found[asString(row["column_name"])] = true
	}
	if !found["Id"] || !found["CamelCase"] {
		t.Fatalf("expected case-sensitive column names, got %#v", found)
	}
}

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

	create := fmt.Sprintf("CREATE TABLE %s.\"MixedCaseTable\" (\"PK\" int PRIMARY KEY, \"Value\" text)", keyspace)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	insert := fmt.Sprintf("INSERT INTO %s.\"MixedCaseTable\" (\"PK\", \"Value\") VALUES (?, ?)", keyspace)
	if err := session.Query(insert, 1, "Test").Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	var value string
	selectStmt := fmt.Sprintf("SELECT \"Value\" FROM %s.MixedCaseTable WHERE \"PK\" = ?", keyspace)
	if err := session.Query(selectStmt, 1).Scan(&value); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if value != "Test" {
		t.Fatalf("unexpected value: %q", value)
	}
}

func TestQuotedLongTableCaseInsensitiveLookup(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 := "Vector_index_f882947b_e67d_4eb3_8c9d_8649fcb115e2_Node"
	create := fmt.Sprintf(`CREATE TABLE %s."%s" (
  id ascii,
  vector text,
  tokens text,
  version bigint,
  tokenizer_version int,
  created_at timestamp,
  updated_at timestamp,
  deleted_at timestamp,
  __tokens__page_content text,
  PRIMARY KEY (id)
)`, 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) VALUES (?)`, keyspace, table)
	if err := session.Query(insert, "row1").Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	var id string
	selectStmt := fmt.Sprintf("SELECT id FROM %s.%s WHERE id = ?", keyspace, table)
	if err := session.Query(selectStmt, "row1").Scan(&id); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if id != "row1" {
		t.Fatalf("unexpected value: %q", id)
	}

	var idLower string
	selectLower := fmt.Sprintf("SELECT id FROM %s.%s WHERE id = ?", keyspace, strings.ToLower(table))
	if err := session.Query(selectLower, "row1").Scan(&idLower); err != nil {
		t.Fatalf("select lower failed: %v", err)
	}
	if idLower != "row1" {
		t.Fatalf("unexpected lower value: %q", idLower)
	}
}