//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"
)

func TestDMLUpdateDeleteWithNonPKAndRange(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 := "chat_items"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (user_id int, session_id int, scenario int, unread_count int, PRIMARY KEY (user_id, session_id))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	createIndex := fmt.Sprintf("CREATE INDEX idx_scenario ON %s.%s (scenario)", keyspace, table)
	if err := session.Query(createIndex).Exec(); err != nil {
		t.Fatalf("create index failed: %v", err)
	}

	insert := fmt.Sprintf("INSERT INTO %s.%s (user_id, session_id, scenario, unread_count) VALUES (?, ?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, 10, 1, 3).Exec(); err != nil {
		t.Fatalf("insert row1 failed: %v", err)
	}
	if err := session.Query(insert, 1, 11, 1, 0).Exec(); err != nil {
		t.Fatalf("insert row2 failed: %v", err)
	}
	if err := session.Query(insert, 1, 12, 2, 5).Exec(); err != nil {
		t.Fatalf("insert row3 failed: %v", err)
	}

	update := fmt.Sprintf(
		"UPDATE %s.%s SET unread_count = unread_count + 1 WHERE user_id = 1 AND scenario = 1 AND unread_count > 0",
		keyspace,
		table,
	)
	if err := session.Query(update).Exec(); err != nil {
		t.Fatalf("update with range failed: %v", err)
	}

	var unread int
	selectRow := fmt.Sprintf("SELECT unread_count FROM %s.%s WHERE user_id = ? AND session_id = ?", keyspace, table)
	if err := session.Query(selectRow, 1, 10).Scan(&unread); err != nil {
		t.Fatalf("select row1 failed: %v", err)
	}
	if unread != 4 {
		t.Fatalf("expected unread_count 4, got %d", unread)
	}
	if err := session.Query(selectRow, 1, 11).Scan(&unread); err != nil {
		t.Fatalf("select row2 failed: %v", err)
	}
	if unread != 0 {
		t.Fatalf("expected unread_count 0, got %d", unread)
	}

	updateIn := fmt.Sprintf(
		"UPDATE %s.%s SET unread_count = 0 WHERE user_id = 1 AND session_id IN (10, 12)",
		keyspace,
		table,
	)
	if err := session.Query(updateIn).Exec(); err != nil {
		t.Fatalf("update with IN failed: %v", err)
	}
	if err := session.Query(selectRow, 1, 10).Scan(&unread); err != nil {
		t.Fatalf("select row1 after IN update failed: %v", err)
	}
	if unread != 0 {
		t.Fatalf("expected unread_count 0 after IN update, got %d", unread)
	}

	updatePkIn := fmt.Sprintf(
		"UPDATE %s.%s SET unread_count = 7 WHERE user_id = 1 AND session_id IN (10)",
		keyspace,
		table,
	)
	if err := session.Query(updatePkIn).Exec(); err != nil {
		t.Fatalf("update with PK IN failed: %v", err)
	}
	if err := session.Query(selectRow, 1, 10).Scan(&unread); err != nil {
		t.Fatalf("select row1 after PK IN update failed: %v", err)
	}
	if unread != 7 {
		t.Fatalf("expected unread_count 7 after PK IN update, got %d", unread)
	}

	deleteStmt := fmt.Sprintf(
		"DELETE FROM %s.%s WHERE user_id = 1 AND scenario = 2 AND session_id >= 12",
		keyspace,
		table,
	)
	if err := session.Query(deleteStmt).Exec(); err != nil {
		t.Fatalf("delete with range failed: %v", err)
	}

	rows, err := scanRows(session, fmt.Sprintf("SELECT session_id FROM %s.%s WHERE user_id = 1 AND session_id = 12", keyspace, table))
	if err != nil {
		t.Fatalf("select deleted row failed: %v", err)
	}
	if len(rows) != 0 {
		t.Fatalf("expected row to be deleted, got %+v", rows)
	}
}