package im

import (
	"context"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

func TestLiveEventEmitterTaskUnreadPublishSequencingIsKeyed(t *testing.T) {
	baseHub := NewStreamEventHub()
	hub := &blockingTaskUnreadKeyHub{
		StreamEventHub: baseHub,
		blockUserID:    "usr_blocked",
		blockScopeID:   "task_blocked",
		firstEntered:   make(chan struct{}),
		releaseFirst:   make(chan struct{}),
	}
	defer func() {
		select {
		case <-hub.releaseFirst:
		default:
			close(hub.releaseFirst)
		}
	}()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	blockedOwner := baseHub.SubscribeStreamEvents(ctx, "wsp_1", "mem_blocked", "usr_blocked")
	unrelatedOwner := baseHub.SubscribeStreamEvents(ctx, "wsp_1", "mem_unrelated", "usr_unrelated")
	defer blockedOwner.Close()
	defer unrelatedOwner.Close()

	emitter := NewLiveEventEmitter(hub, nil, nil)
	firstDone := make(chan struct{})
	go func() {
		defer close(firstDone)
		emitter.EmitUnreadUpdated("wsp_1", taskUnreadDelta(
			"usr_blocked", "mem_blocked", "task_blocked", 1,
		))
	}()
	<-hub.firstEntered

	sameKeyDone := make(chan struct{})
	go func() {
		defer close(sameKeyDone)
		emitter.EmitUnreadUpdated("wsp_1", taskUnreadDelta(
			"usr_blocked", "mem_blocked", "task_blocked", 2,
		))
	}()
	select {
	case <-sameKeyDone:
		t.Fatal("a later Task unread publish passed the blocked publish for the same owner and scope")
	case <-time.After(50 * time.Millisecond):
	}

	otherTaskDone := make(chan struct{})
	go func() {
		defer close(otherTaskDone)
		emitter.EmitUnreadUpdated("wsp_1", taskUnreadDelta(
			"usr_blocked", "mem_blocked", "task_other", 6,
		))
	}()
	select {
	case <-otherTaskDone:
	case <-time.After(time.Second):
		t.Fatal("another Task for the same owner was blocked by the first Task")
	}
	otherTask := receiveHubEvent(t, blockedOwner)
	require.Equal(t, "task_other", otherTask.ScopeID)
	require.Equal(t, int64(6), otherTask.Payload["count"])

	unrelatedDone := make(chan struct{})
	go func() {
		defer close(unrelatedDone)
		emitter.EmitUnreadUpdated("wsp_1", taskUnreadDelta(
			"usr_unrelated", "mem_unrelated", "task_blocked", 7,
		))
	}()
	select {
	case <-unrelatedDone:
	case <-time.After(time.Second):
		t.Fatal("another owner of the same Task was blocked by the first owner")
	}
	unrelated := receiveHubEvent(t, unrelatedOwner)
	require.Equal(t, int64(7), unrelated.Payload["count"])

	close(hub.releaseFirst)
	<-firstDone
	<-sameKeyDone

	first := receiveHubEvent(t, blockedOwner)
	second := receiveHubEvent(t, blockedOwner)
	require.Equal(t, int64(1), first.Payload["count"])
	require.Equal(t, int64(2), second.Payload["count"])

	emitter.taskUnreadPublishSequencer.mu.Lock()
	activeKeys := len(emitter.taskUnreadPublishSequencer.entries)
	emitter.taskUnreadPublishSequencer.mu.Unlock()
	require.Zero(t, activeKeys, "idle Task keys must not accumulate in emitter memory")
}

func taskUnreadDelta(userID, memberID, scopeID string, count int64) unread.UnreadDelta {
	return unread.UnreadDelta{
		WorkspaceID:  "wsp_1",
		UserID:       userID,
		MemberID:     memberID,
		ScopeID:      scopeID,
		ScopeType:    unread.ScopeTypeTask,
		Count:        count,
		HasUnread:    count > 0,
		ServingState: unread.ServingStateReady,
	}
}

type blockingTaskUnreadKeyHub struct {
	*StreamEventHub

	blockUserID  string
	blockScopeID string
	mu           sync.Mutex
	blockCalls   int
	firstEntered chan struct{}
	releaseFirst chan struct{}
}

func (hub *blockingTaskUnreadKeyHub) PublishUserLiveOnlyEvent(workspaceID string, userID string, event storeim.StreamEvent) {
	if userID == hub.blockUserID && event.ScopeID == hub.blockScopeID {
		hub.mu.Lock()
		hub.blockCalls++
		call := hub.blockCalls
		hub.mu.Unlock()
		if call == 1 {
			close(hub.firstEntered)
			<-hub.releaseFirst
		}
	}
	hub.StreamEventHub.PublishUserLiveOnlyEvent(workspaceID, userID, event)
}