package im

import (
	"sync"
	"sync/atomic"
	"testing"

	"github.com/stretchr/testify/require"

	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

func TestUnreadProjectionReplayPayloadsPreserveScopeTypes(t *testing.T) {
	t.Parallel()

	reader := replayUnreadProjectionReader{
		states: map[string]unread.ScopeUnreadState{
			"chn_1": {
				WorkspaceID:        "wsp_1",
				MemberID:           "mem_1",
				ScopeID:            "chn_1",
				ScopeType:          unread.ScopeTypeChannel,
				MessageUnreadCount: 1,
				ServingState:       unread.ServingStateReady,
			},
			"task_scope_1": {
				WorkspaceID:         "wsp_1",
				MemberID:            "mem_1",
				ScopeID:             "task_scope_1",
				ScopeType:           unread.ScopeTypeTask,
				ActivityUnreadCount: 2,
				ServingState:        unread.ServingStateReady,
			},
		},
		workspaceUnreadConversationCount: 99,
	}

	payloads := UnreadProjectionReplayPayloads("wsp_1", "mem_1", reader)

	require.Len(t, payloads, 2)
	require.Equal(t, "channel", payloads[0]["scopeType"])
	require.Equal(t, int64(1), payloads[0]["count"])
	require.Equal(t, "task", payloads[1]["scopeType"])
	require.Equal(t, int64(2), payloads[1]["count"])
	require.Equal(t, int64(2), payloads[1]["workspaceUnreadConversationCount"])
}

func TestUnreadProjectionReplayPayloadsDeriveAggregateFromCapturedScopeSnapshot(t *testing.T) {
	snapshotCaptured := make(chan struct{})
	concurrentChangeApplied := make(chan struct{})
	reader := &changingReplayUnreadProjectionReader{
		states: map[string]unread.ScopeUnreadState{
			"task_scope_1": {
				WorkspaceID:  "wsp_1",
				MemberID:     "mem_1",
				ScopeID:      "task_scope_1",
				ScopeType:    unread.ScopeTypeTask,
				ServingState: unread.ServingStateReady,
			},
		},
		snapshotCaptured:        snapshotCaptured,
		concurrentChangeApplied: concurrentChangeApplied,
	}

	var change sync.WaitGroup
	change.Add(1)
	go func() {
		defer change.Done()
		<-snapshotCaptured
		// Model one scope becoming unread after ScopeStates captured the
		// bootstrap snapshot. Its live delta is buffered by the subscription
		// until after bootstrap_complete.
		reader.currentWorkspaceUnreadConversationCount.Store(1)
		close(concurrentChangeApplied)
	}()

	payloads := UnreadProjectionReplayPayloads("wsp_1", "mem_1", reader)
	change.Wait()

	require.Len(t, payloads, 1)
	require.Equal(t, int64(0), payloads[0]["count"])
	require.Equal(t, false, payloads[0]["hasUnread"])
	require.Equal(t, false, payloads[0]["workspaceHasUnread"])
	require.Equal(
		t,
		int64(0),
		payloads[0]["workspaceUnreadConversationCount"],
		"the authoritative aggregate must describe the captured scope snapshot",
	)
	require.Zero(t, reader.aggregateReadCalls.Load(), "bootstrap must not perform a second aggregate read")

	// Apply the bootstrap payload followed by the concurrently buffered
	// scope-local delta using the browser's transition semantics. Starting from
	// the captured aggregate (0) makes the one false->true transition converge
	// to the real total (1), rather than double-counting it as 2.
	accumulator := unreadConversationAccumulator{
		scopeHasUnread: map[string]bool{},
	}
	accumulator.apply(payloads[0], true)
	accumulator.apply(map[string]any{
		"scopeId":   "task_scope_1",
		"hasUnread": true,
	}, false)
	require.Equal(t, int64(1), accumulator.count)
	require.True(t, accumulator.scopeHasUnread["task_scope_1"])
}

func TestReplayUnreadProjectionStateStartsWithAuthoritativeReset(t *testing.T) {
	t.Parallel()

	emitter := &LiveEventEmitter{}
	var events []storeim.StreamEvent
	emit := func(event storeim.StreamEvent) {
		events = append(events, event)
	}

	emitter.ReplayUnreadProjectionState(
		" wsp_1 ",
		" mem_1 ",
		replayUnreadProjectionReader{},
		emit,
	)

	require.Len(t, events, 1, "an empty authoritative snapshot must still clear the previous reconnect epoch")
	require.Equal(t, liveEventTypeUnreadReset, events[0].Type)
	require.Equal(t, "wsp_1", events[0].WorkspaceID)
	require.Equal(t, "mem_1", events[0].MemberID)
	require.Equal(t, map[string]any{
		"workspaceId": "wsp_1",
		"memberId":    "mem_1",
	}, events[0].Payload)

	events = nil
	emitter.ReplayUnreadProjectionState(
		"wsp_1",
		"mem_1",
		replayUnreadProjectionReader{states: map[string]unread.ScopeUnreadState{
			"task_scope_1": {
				WorkspaceID:         "wsp_1",
				MemberID:            "mem_1",
				ScopeID:             "task_scope_1",
				ScopeType:           unread.ScopeTypeTask,
				ActivityUnreadCount: 1,
				ServingState:        unread.ServingStateReady,
			},
		}},
		emit,
	)

	require.Len(t, events, 2)
	require.Equal(t, liveEventTypeUnreadReset, events[0].Type)
	require.Equal(t, liveEventTypeUnreadUpdated, events[1].Type)
	require.Equal(t, "task_scope_1", events[1].ScopeID)
	require.Equal(t, int64(1), events[1].Payload["workspaceUnreadConversationCount"])
}

type replayUnreadProjectionReader struct {
	states                           map[string]unread.ScopeUnreadState
	workspaceUnreadConversationCount int64
}

func (reader replayUnreadProjectionReader) ScopeStates(string, string, []string) map[string]unread.ScopeUnreadState {
	return reader.states
}

func (reader replayUnreadProjectionReader) WorkspaceHasUnread(string, string) bool {
	return reader.workspaceUnreadConversationCount > 0
}

func (reader replayUnreadProjectionReader) WorkspaceUnreadConversationCount(string, string) int64 {
	return reader.workspaceUnreadConversationCount
}

type changingReplayUnreadProjectionReader struct {
	states                                  map[string]unread.ScopeUnreadState
	snapshotCaptured                        chan struct{}
	concurrentChangeApplied                 chan struct{}
	currentWorkspaceUnreadConversationCount atomic.Int64
	aggregateReadCalls                      atomic.Int64
}

func (reader *changingReplayUnreadProjectionReader) ScopeStates(string, string, []string) map[string]unread.ScopeUnreadState {
	snapshot := make(map[string]unread.ScopeUnreadState, len(reader.states))
	for scopeID, state := range reader.states {
		snapshot[scopeID] = state
	}
	close(reader.snapshotCaptured)
	<-reader.concurrentChangeApplied
	return snapshot
}

// These methods deliberately remain on the fake even though the production
// reader contract no longer requires them. The pre-fix implementation called
// them after ScopeStates and observed the concurrent value, making this test
// report an aggregate of 1 for a snapshot whose only scope was still read.
func (reader *changingReplayUnreadProjectionReader) WorkspaceHasUnread(string, string) bool {
	reader.aggregateReadCalls.Add(1)
	return reader.currentWorkspaceUnreadConversationCount.Load() > 0
}

func (reader *changingReplayUnreadProjectionReader) WorkspaceUnreadConversationCount(string, string) int64 {
	reader.aggregateReadCalls.Add(1)
	return reader.currentWorkspaceUnreadConversationCount.Load()
}

type unreadConversationAccumulator struct {
	count          int64
	scopeHasUnread map[string]bool
}

func (accumulator *unreadConversationAccumulator) apply(payload map[string]any, authoritative bool) {
	scopeID, _ := payload["scopeId"].(string)
	hasUnread, _ := payload["hasUnread"].(bool)
	previous := accumulator.scopeHasUnread[scopeID]
	accumulator.scopeHasUnread[scopeID] = hasUnread
	if authoritative {
		accumulator.count, _ = payload["workspaceUnreadConversationCount"].(int64)
		return
	}
	if previous == hasUnread {
		return
	}
	if hasUnread {
		accumulator.count++
		return
	}
	if accumulator.count > 0 {
		accumulator.count--
	}
}