//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"strconv"
	"testing"

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

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (user_id, session_id, note) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, 1, "seed").Exec(); err != nil {
		t.Fatalf("insert seed row failed: %v", err)
	}

	updateFullPK := fmt.Sprintf(
		"UPDATE %s.%s SET unread_count = unread_count + 1 WHERE user_id = ? AND session_id = ?",
		keyspace,
		table,
	)
	if err := session.Query(updateFullPK, 1, 1).Exec(); err != nil {
		t.Fatalf("update full pk on null row failed: %v", err)
	}
	if err := session.Query(updateFullPK, 1, 1).Exec(); err != nil {
		t.Fatalf("second update full pk failed: %v", err)
	}

	row1, err := singleRow(session, fmt.Sprintf("SELECT unread_count FROM %s.%s WHERE user_id = 1 AND session_id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select updated row1 failed: %v", err)
	}
	unread1, err := parseIntField(row1["unread_count"])
	if err != nil {
		t.Fatalf("parse row1 unread_count failed: %v", err)
	}
	if unread1 != 2 {
		t.Fatalf("expected row1 unread_count 2, got %d", unread1)
	}

	if err := session.Query(updateFullPK, 2, 1).Exec(); err != nil {
		t.Fatalf("update full pk for missing row failed: %v", err)
	}
	row2, err := singleRow(session, fmt.Sprintf("SELECT unread_count FROM %s.%s WHERE user_id = 2 AND session_id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select upserted row2 failed: %v", err)
	}
	unread2, err := parseIntField(row2["unread_count"])
	if err != nil {
		t.Fatalf("parse row2 unread_count failed: %v", err)
	}
	if unread2 != 1 {
		t.Fatalf("expected row2 unread_count 1, got %d", unread2)
	}

	if err := session.Query(insert, 3, 1, "range").Exec(); err != nil {
		t.Fatalf("insert range row failed: %v", err)
	}
	updateRange := fmt.Sprintf(
		"UPDATE %s.%s SET unread_count = unread_count + 1 WHERE user_id = 3 AND session_id > 0",
		keyspace,
		table,
	)
	if err := session.Query(updateRange).Exec(); err != nil {
		t.Fatalf("range update failed: %v", err)
	}
	row3, err := singleRow(session, fmt.Sprintf("SELECT note FROM %s.%s WHERE user_id = 3 AND session_id = 1", keyspace, table))
	if err != nil {
		t.Fatalf("select range row failed: %v", err)
	}
	if asString(row3["note"]) != "range" {
		t.Fatalf("unexpected existing row after range update: %#v", row3)
	}

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

	rowsMissing, err := scanRows(session, fmt.Sprintf("SELECT session_id FROM %s.%s WHERE user_id = 4", keyspace, table))
	if err != nil {
		t.Fatalf("select missing rows after range update failed: %v", err)
	}
	if len(rowsMissing) != 0 {
		t.Fatalf("expected no upsert rows for range update, got %#v", rowsMissing)
	}
}

func singleRow(session *gocql.Session, query string) (map[string]interface{}, error) {
	rows, err := scanRows(session, query)
	if err != nil {
		return nil, err
	}
	if len(rows) != 1 {
		return nil, fmt.Errorf("expected exactly 1 row, got %d", len(rows))
	}
	return rows[0], nil
}

func parseIntField(value interface{}) (int64, error) {
	if value == nil {
		return 0, fmt.Errorf("value is NULL")
	}
	return strconv.ParseInt(asString(value), 10, 64)
}