//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
)

func TestConcurrentArithmeticUpdatesMatchShadowModel(t *testing.T) {
	session := newSession(t, sessionOptions{
		disableHostLookup: true,
		numConns:          4,
	})

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

	const (
		keyCount            = 16
		workers             = 8
		iterationsPerWorker = 40
		userID              = 7
	)

	insert := fmt.Sprintf(
		"INSERT INTO %s.%s (user_id, session_id, unread_count, mention_count, seq_no) VALUES (?, ?, ?, ?, ?)",
		keyspace,
		table,
	)
	for sessionID := 0; sessionID < keyCount; sessionID++ {
		if err := session.Query(insert, userID, sessionID, 0, 0, int64(0)).Exec(); err != nil {
			t.Fatalf("insert seed row %d failed: %v", sessionID, err)
		}
	}

	update := fmt.Sprintf(
		"UPDATE %s.%s SET unread_count = unread_count + ?, mention_count = mention_count + ?, seq_no = seq_no + ? WHERE user_id = ? AND session_id = ?",
		keyspace,
		table,
	)

	var expectedUnread [keyCount]atomic.Int64
	var expectedMention [keyCount]atomic.Int64
	var expectedSeq [keyCount]atomic.Int64

	var wg sync.WaitGroup
	errCh := make(chan error, workers)
	for worker := 0; worker < workers; worker++ {
		wg.Add(1)
		go func(worker int) {
			defer wg.Done()
			for iter := 0; iter < iterationsPerWorker; iter++ {
				sessionID := (worker*3 + iter*5) % keyCount
				unreadDelta := int64(1 + (worker+iter)%3)
				mentionDelta := int64((worker + 2*iter) % 2)
				seqDelta := int64(1)

				if err := session.Query(update, unreadDelta, mentionDelta, seqDelta, userID, sessionID).Exec(); err != nil {
					errCh <- fmt.Errorf("worker=%d iter=%d sessionID=%d update failed: %w", worker, iter, sessionID, err)
					return
				}

				expectedUnread[sessionID].Add(unreadDelta)
				expectedMention[sessionID].Add(mentionDelta)
				expectedSeq[sessionID].Add(seqDelta)
			}
		}(worker)
	}

	wg.Wait()
	close(errCh)
	for err := range errCh {
		if err != nil {
			t.Fatal(err)
		}
	}

	var totalUnread int64
	var totalMention int64
	var totalSeq int64
	for sessionID := 0; sessionID < keyCount; sessionID++ {
		row, err := singleRow(session, fmt.Sprintf(
			"SELECT unread_count, mention_count, seq_no FROM %s.%s WHERE user_id = %d AND session_id = %d",
			keyspace,
			table,
			userID,
			sessionID,
		))
		if err != nil {
			t.Fatalf("select sessionID=%d failed: %v", sessionID, err)
		}

		unread, err := parseIntField(row["unread_count"])
		if err != nil {
			t.Fatalf("parse unread_count for sessionID=%d failed: %v", sessionID, err)
		}
		mention, err := parseIntField(row["mention_count"])
		if err != nil {
			t.Fatalf("parse mention_count for sessionID=%d failed: %v", sessionID, err)
		}
		seq, err := parseIntField(row["seq_no"])
		if err != nil {
			t.Fatalf("parse seq_no for sessionID=%d failed: %v", sessionID, err)
		}

		wantUnread := expectedUnread[sessionID].Load()
		wantMention := expectedMention[sessionID].Load()
		wantSeq := expectedSeq[sessionID].Load()
		if unread != wantUnread || mention != wantMention || seq != wantSeq {
			t.Fatalf(
				"shadow mismatch for sessionID=%d: got unread=%d mention=%d seq=%d, want unread=%d mention=%d seq=%d",
				sessionID,
				unread,
				mention,
				seq,
				wantUnread,
				wantMention,
				wantSeq,
			)
		}

		totalUnread += wantUnread
		totalMention += wantMention
		totalSeq += wantSeq
	}

	aggregate, err := singleRow(session, fmt.Sprintf(
		"SELECT COUNT(*) AS cnt, SUM(unread_count) AS sum_unread, SUM(mention_count) AS sum_mention, SUM(seq_no) AS sum_seq FROM %s.%s WHERE user_id = %d",
		keyspace,
		table,
		userID,
	))
	if err != nil {
		t.Fatalf("select aggregate failed: %v", err)
	}
	count, err := parseIntField(aggregate["cnt"])
	if err != nil {
		t.Fatalf("parse aggregate count failed: %v", err)
	}
	sumUnread, err := parseIntField(aggregate["sum_unread"])
	if err != nil {
		t.Fatalf("parse aggregate sum_unread failed: %v", err)
	}
	sumMention, err := parseIntField(aggregate["sum_mention"])
	if err != nil {
		t.Fatalf("parse aggregate sum_mention failed: %v", err)
	}
	sumSeq, err := parseIntField(aggregate["sum_seq"])
	if err != nil {
		t.Fatalf("parse aggregate sum_seq failed: %v", err)
	}

	if count != keyCount || sumUnread != totalUnread || sumMention != totalMention || sumSeq != totalSeq {
		t.Fatalf(
			"aggregate shadow mismatch: got count=%d unread=%d mention=%d seq=%d, want count=%d unread=%d mention=%d seq=%d",
			count,
			sumUnread,
			sumMention,
			sumSeq,
			keyCount,
			totalUnread,
			totalMention,
			totalSeq,
		)
	}
}