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 TestStreamEventHubPublishesOnlyMatchingMember(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	defer owner.Close()
	defer peer.Close()

	hub.PublishStreamEvents([]storeim.StreamEvent{
		{WorkspaceID: "wsp_1", MemberID: "mem_owner", EventSeq: 1, EventID: "evt_1", Type: "message.created"},
	})

	require.Equal(t, int64(1), receiveHubEvent(t, owner).EventSeq)
	requireNoHubEvent(t, peer)
}

func TestLiveEventEmitterBroadcastsServiceAccountDeletionToWorkspace(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	otherWorkspace := hub.SubscribeStreamEvents(ctx, "wsp_2", "mem_other", "usr_other")
	defer owner.Close()
	defer peer.Close()
	defer otherWorkspace.Close()

	NewLiveEventEmitter(hub, nil, nil).EmitServiceAccountDeleted(
		" wsp_1 ", " sacct_1 ", " svc_1 ", " sasc_1 ",
	)

	for _, subscription := range []*StreamEventSubscription{owner, peer} {
		event := receiveHubEvent(t, subscription)
		require.Equal(t, "runtime.service_account_deleted", event.Type)
		require.Equal(t, "wsp_1", event.WorkspaceID)
		require.Equal(t, "sasc_1", event.ScopeID)
		require.Zero(t, event.EventSeq)
		require.Equal(t, map[string]any{
			"workspaceId": "wsp_1",
			"accountId":   "sacct_1",
			"principalId": "svc_1",
			"scopeId":     "sasc_1",
		}, event.Payload)
	}
	requireNoHubEvent(t, otherWorkspace)
}

func TestLiveEventEmitterBroadcastsWorkspaceMemberRoleUpdate(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	defer owner.Close()
	defer peer.Close()

	NewLiveEventEmitter(hub, nil, nil).EmitWorkspaceMemberRoleUpdated(
		" wsp_1 ", " usr_target ", " admin ",
	)

	for _, subscription := range []*StreamEventSubscription{owner, peer} {
		event := receiveHubEvent(t, subscription)
		require.Equal(t, "runtime.workspace_members_updated", event.Type)
		require.Equal(t, "wsp_1", event.WorkspaceID)
		require.Zero(t, event.EventSeq)
		require.Equal(t, map[string]any{
			"workspaceId":  "wsp_1",
			"changeKind":   "role_updated",
			"targetUserId": "usr_target",
			"role":         "admin",
		}, event.Payload)
	}
}

func TestLiveEventEmitterBroadcastsWorkspaceAppUpdate(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	otherWorkspace := hub.SubscribeStreamEvents(ctx, "wsp_2", "mem_other", "usr_other")
	defer owner.Close()
	defer peer.Close()
	defer otherWorkspace.Close()

	NewLiveEventEmitter(hub, nil, nil).EmitWorkspaceAppsUpdated(" wsp_1 ")

	for _, subscription := range []*StreamEventSubscription{owner, peer} {
		event := receiveHubEvent(t, subscription)
		require.Equal(t, "runtime.workspace_apps_updated", event.Type)
		require.Equal(t, "wsp_1", event.WorkspaceID)
		require.Zero(t, event.EventSeq)
		require.Equal(t, map[string]any{"workspaceId": "wsp_1"}, event.Payload)
	}
	requireNoHubEvent(t, otherWorkspace)
}

func TestLiveEventEmitterIncludesUnreadScopeType(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	defer owner.Close()

	NewLiveEventEmitter(hub, nil, nil).EmitUnreadUpdated("wsp_1", unread.UnreadDelta{
		WorkspaceID:  "wsp_1",
		UserID:       "usr_owner",
		MemberID:     "mem_owner",
		ScopeID:      "custom_task_scope",
		ScopeType:    unread.ScopeTypeTask,
		Count:        4,
		HasUnread:    true,
		ServingState: unread.ServingStateReady,
	})

	event := receiveHubEvent(t, owner)
	require.Equal(t, "runtime.unread_updated", event.Type)
	require.Equal(t, "task", event.Payload["scopeType"])
	require.Equal(t, int64(4), event.Payload["count"])
	require.NotContains(t, event.Payload, "workspaceUnreadConversationCount", "scope-local Task live updates must never overwrite the cross-scope aggregate")

	NewLiveEventEmitter(hub, nil, nil).EmitUnreadUpdated("wsp_1", unread.UnreadDelta{
		WorkspaceID:                      "wsp_1",
		UserID:                           "usr_owner",
		MemberID:                         "mem_owner",
		ScopeID:                          "chn_1",
		ScopeType:                        unread.ScopeTypeChannel,
		Count:                            1,
		HasUnread:                        true,
		WorkspaceUnreadConversationCount: 99,
		ServingState:                     unread.ServingStateReady,
	})

	event = receiveHubEvent(t, owner)
	require.Equal(t, "channel", event.Payload["scopeType"])
	require.Equal(t, int64(1), event.Payload["count"])
	require.NotContains(t, event.Payload, "workspaceUnreadConversationCount", "no scope-local live update can own a cross-scope aggregate")
}

func TestLiveEventEmitterReplacesDelayedOlderTaskUnreadProjectionWithCurrentTruth(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	defer owner.Close()

	runtime, err := unread.New(unread.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster("wsp_1", "task_scope_1", unread.ScopeTypeTask, []unread.RosterMember{{
		MemberID:          "mem_owner",
		UserID:            "usr_owner",
		Visibility:        "public",
		NotificationState: "active",
	}}))
	applied, err := runtime.ApplyMessage(unread.MessageStored{
		WorkspaceID:        "wsp_1",
		ScopeID:            "task_scope_1",
		ScopeType:          unread.ScopeTypeTask,
		Seq:                1,
		RecipientMemberIDs: []string{"mem_owner"},
	})
	require.NoError(t, err)
	require.Len(t, applied.Deltas, 1)

	emitter := NewLiveEventEmitter(hub, nil, nil)
	emitter.SetUnreadCurrentDeltaReader(runtime)
	emitter.EmitUnreadUpdated("wsp_1", applied.Deltas[0])
	emitter.EmitUnreadUpdated("wsp_1", unread.UnreadDelta{
		WorkspaceID:  "wsp_1",
		UserID:       "usr_owner",
		MemberID:     "mem_owner",
		ScopeID:      "task_scope_1",
		ScopeType:    unread.ScopeTypeTask,
		ServingState: unread.ServingStateReady,
		StateVersion: 1,
	})

	for range 2 {
		event := receiveHubEvent(t, owner)
		require.Equal(t, "runtime.unread_updated", event.Type)
		require.Equal(t, int64(1), event.Payload["count"])
		require.Equal(t, true, event.Payload["hasUnread"])
	}
}

func TestLiveEventEmitterSerializesTaskUnreadRereadAndPublish(t *testing.T) {
	baseHub := NewStreamEventHub()
	hub := &blockingFirstUserLiveHub{
		StreamEventHub: baseHub,
		firstEntered:   make(chan struct{}),
		releaseFirst:   make(chan struct{}),
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := baseHub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	defer owner.Close()

	runtime, err := unread.New(unread.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster("wsp_1", "task_scope_1", unread.ScopeTypeTask, []unread.RosterMember{{
		MemberID:          "mem_owner",
		UserID:            "usr_owner",
		Visibility:        "public",
		NotificationState: "active",
	}}))
	applied, err := runtime.ApplyMessage(unread.MessageStored{
		WorkspaceID:        "wsp_1",
		ScopeID:            "task_scope_1",
		ScopeType:          unread.ScopeTypeTask,
		Seq:                1,
		RecipientMemberIDs: []string{"mem_owner"},
	})
	require.NoError(t, err)
	require.Len(t, applied.Deltas, 1)

	emitter := NewLiveEventEmitter(hub, nil, nil)
	emitter.SetUnreadCurrentDeltaReader(runtime)
	firstDone := make(chan struct{})
	go func() {
		defer close(firstDone)
		emitter.EmitUnreadUpdated("wsp_1", applied.Deltas[0])
	}()
	<-hub.firstEntered

	require.NoError(t, runtime.ApplyReadCursor(unread.ReadCursorAdvanced{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_owner",
		ScopeID:     "task_scope_1",
		LastReadSeq: 1,
	}))
	current, found := runtime.CurrentUnreadDelta(
		"wsp_1",
		"usr_owner",
		"mem_owner",
		"task_scope_1",
		unread.ScopeTypeTask,
	)
	require.True(t, found)
	require.Zero(t, current.Count)

	secondStarted := make(chan struct{})
	secondDone := make(chan struct{})
	go func() {
		close(secondStarted)
		defer close(secondDone)
		emitter.EmitUnreadUpdated("wsp_1", current)
	}()
	<-secondStarted
	select {
	case <-secondDone:
		t.Fatal("newer Task unread publish passed the older blocked publish")
	case <-time.After(50 * time.Millisecond):
	}

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

	first := receiveHubEvent(t, owner)
	second := receiveHubEvent(t, owner)
	require.Equal(t, int64(1), first.Payload["count"])
	require.Equal(t, int64(0), second.Payload["count"])
	requireNoHubEvent(t, owner)
}

type blockingFirstUserLiveHub struct {
	*StreamEventHub

	mu           sync.Mutex
	calls        int
	firstEntered chan struct{}
	releaseFirst chan struct{}
}

func (hub *blockingFirstUserLiveHub) PublishUserLiveOnlyEvent(workspaceID string, userID string, event storeim.StreamEvent) {
	hub.mu.Lock()
	hub.calls++
	call := hub.calls
	hub.mu.Unlock()
	if call == 1 {
		close(hub.firstEntered)
		<-hub.releaseFirst
	}
	hub.StreamEventHub.PublishUserLiveOnlyEvent(workspaceID, userID, event)
}

func TestStreamEventHubPublishesVisibleScopeToNonMemberSubscriberOnce(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	viewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_viewer", "usr_viewer")
	defer owner.Close()
	defer viewer.Close()
	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "thr_1"))

	events := []storeim.StreamEvent{
		{WorkspaceID: "wsp_1", MemberID: "mem_owner", EventSeq: 4, EventID: "evt_thread_1", Type: "message.created", ScopeID: "thr_1", EntityID: "msg_1", Payload: map[string]any{"messageId": "msg_1"}},
		{WorkspaceID: "wsp_1", MemberID: "mem_peer", EventSeq: 9, EventID: "evt_thread_1", Type: "message.created", ScopeID: "thr_1", EntityID: "msg_1", Payload: map[string]any{"messageId": "msg_1"}},
	}
	hub.PublishStreamEvents(events)
	hub.PublishVisibleScopeEvents(events)

	require.Equal(t, int64(4), receiveHubEvent(t, owner).EventSeq)
	visibleEvent := receiveHubEvent(t, viewer)
	require.Equal(t, "mem_viewer", visibleEvent.MemberID)
	require.Equal(t, int64(0), visibleEvent.EventSeq)
	require.Empty(t, visibleEvent.EventID)
	require.Equal(t, "message.created", visibleEvent.Type)
	require.Equal(t, "thr_1", visibleEvent.ScopeID)
	require.Equal(t, map[string]any{"messageId": "msg_1"}, visibleEvent.Payload)
	requireNoHubEvent(t, viewer)
}

func TestStreamEventHubVisibleScopeSkipsDurableMemberDuplicate(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	defer owner.Close()
	require.True(t, hub.SubscribeVisibleScopeEvents(owner, "wsp_1", "mem_owner", "thr_1"))

	events := []storeim.StreamEvent{
		{WorkspaceID: "wsp_1", MemberID: "mem_owner", EventSeq: 4, EventID: "evt_thread_1", Type: "message.created", ScopeID: "thr_1"},
	}
	hub.PublishStreamEvents(events)
	hub.PublishVisibleScopeEvents(events)

	require.Equal(t, int64(4), receiveHubEvent(t, owner).EventSeq)
	requireNoHubEvent(t, owner)
}

func TestStreamEventHubUnsubscribeVisibleScopeStopsLiveDelivery(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	viewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_viewer", "usr_viewer")
	defer viewer.Close()
	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "thr_1"))
	hub.UnsubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "thr_1")

	hub.PublishVisibleScopeEvents([]storeim.StreamEvent{
		{WorkspaceID: "wsp_1", MemberID: "mem_owner", EventSeq: 4, EventID: "evt_thread_1", Type: "message.created", ScopeID: "thr_1"},
	})

	requireNoHubEvent(t, viewer)
}

func TestStreamEventHubExplicitReleasePreservesImplicitScopeSource(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	viewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_viewer", "usr_viewer")
	defer viewer.Close()
	require.True(t, hub.SubscribeImplicitVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "chn_1"))
	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "chn_1"))

	hub.UnsubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "chn_1")
	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_1", storeim.StreamEvent{Type: "runtime.task_updated"})
	require.Equal(t, "runtime.task_updated", receiveHubEvent(t, viewer).Type)

	// Membership removal is an authorization boundary and therefore clears
	// both implicit and explicit sources for every connection owned by the user.
	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "chn_1"))
	hub.UnsubscribeScopeForUser("wsp_1", "usr_viewer", "chn_1")
	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_1", storeim.StreamEvent{Type: "runtime.task_updated"})
	requireNoHubEvent(t, viewer)
}

func TestPublishingStreamEventWriterPublishesPersistedEvents(t *testing.T) {
	base := &fakeStreamEventWriter{scopeEvents: []storeim.StreamEvent{
		{WorkspaceID: "wsp_1", MemberID: "mem_1", EventSeq: 7, EventID: "evt_7", Type: "message.created"},
	}}
	publisher := &recordingStreamEventPublisher{}
	writer := NewPublishingStreamEventWriter(base, publisher)

	events, err := writer.AppendScopeEvent(context.Background(), storeim.ScopeEventRequest{
		WorkspaceID: "wsp_1",
		ScopeID:     "chn_1",
		Event:       storeim.StreamEvent{Type: "message.created"},
	})
	require.NoError(t, err)
	require.Equal(t, base.scopeEvents, events)
	require.Equal(t, base.scopeEvents, publisher.events)
	require.Equal(t, base.scopeEvents, publisher.visibleEvents)
}

func TestPublishingStreamEventWriterForwardsScopeLiveOnlyWithoutDurableAppend(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	viewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_viewer", "usr_viewer")
	defer viewer.Close()
	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "wfs_1"))

	base := &fakeStreamEventWriter{}
	writer := NewPublishingStreamEventWriter(base, hub)
	liveWriter, ok := writer.(interface {
		PublishScopeLiveOnlyEvent(string, string, storeim.StreamEvent)
	})
	require.True(t, ok)
	liveWriter.PublishScopeLiveOnlyEvent("wsp_1", "wfs_1", storeim.StreamEvent{
		Type:     "message.created",
		EntityID: "msg_workflow_live",
		Payload:  map[string]any{"messageId": "msg_workflow_live"},
	})

	event := receiveHubEvent(t, viewer)
	require.Equal(t, "message.created", event.Type)
	require.Equal(t, "wfs_1", event.ScopeID)
	require.Equal(t, "mem_viewer", event.MemberID)
	require.Zero(t, event.EventSeq)
	require.Empty(t, event.EventID)
	require.Zero(t, base.scopeAppendCalls)
}

func receiveHubEvent(t *testing.T, subscription *StreamEventSubscription) storeim.StreamEvent {
	t.Helper()
	select {
	case event := <-subscription.Events():
		return event
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for hub event")
		return storeim.StreamEvent{}
	}
}

func requireNoHubEvent(t *testing.T, subscription *StreamEventSubscription) {
	t.Helper()
	select {
	case event := <-subscription.Events():
		t.Fatalf("unexpected hub event: %+v", event)
	case <-time.After(50 * time.Millisecond):
	}
}

func TestStreamEventHubLiveOnlyByMemberStripsEventSeq(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	defer owner.Close()

	hub.PublishLiveOnlyEventsByMember([]storeim.StreamEvent{{
		WorkspaceID: "wsp_1", MemberID: "mem_owner",
		EventSeq: 99, EventID: "evt_x", CursorGeneration: "cur_x",
		Type: "runtime.unread_updated",
	}})

	event := receiveHubEvent(t, owner)
	require.Equal(t, int64(0), event.EventSeq)
	require.Empty(t, event.EventID)
	require.Empty(t, event.CursorGeneration)
	require.Equal(t, "runtime.unread_updated", event.Type)
}

func TestStreamEventHubWorkspaceBroadcastReachesAllMembers(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	outside := hub.SubscribeStreamEvents(ctx, "wsp_2", "mem_other", "usr_other")
	defer owner.Close()
	defer peer.Close()
	defer outside.Close()

	hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: "wsp_1", EventSeq: 7, EventID: "evt_7",
		Type: "runtime.agent_status",
	})

	for _, sub := range []*StreamEventSubscription{owner, peer} {
		event := receiveHubEvent(t, sub)
		require.Equal(t, int64(0), event.EventSeq)
		require.Empty(t, event.EventID)
		require.Equal(t, "runtime.agent_status", event.Type)
	}
	requireNoHubEvent(t, outside)
}

func TestStreamEventHubUserLiveOnlyReachesOnlyOwner(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	owner := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_owner", "usr_owner")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	defer owner.Close()
	defer peer.Close()

	hub.PublishUserLiveOnlyEvent("wsp_1", "usr_owner", storeim.StreamEvent{
		Type:    "runtime.agent_activity",
		Payload: map[string]any{"agentId": "agt_1"},
	})

	event := receiveHubEvent(t, owner)
	require.Equal(t, "runtime.agent_activity", event.Type)
	require.Equal(t, "mem_owner", event.MemberID)
	require.Equal(t, int64(0), event.EventSeq)
	requireNoHubEvent(t, peer)
}

func TestStreamEventHubScopeLiveOnlyReachesOnlyScopeSubscriber(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	viewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_viewer", "usr_viewer")
	otherChannelViewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_other_channel", "usr_other_channel")
	workspacePeer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	outsideWorkspace := hub.SubscribeStreamEvents(ctx, "wsp_2", "mem_other", "usr_other")
	defer viewer.Close()
	defer otherChannelViewer.Close()
	defer workspacePeer.Close()
	defer outsideWorkspace.Close()

	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "chn_alpha"))
	require.True(t, hub.SubscribeVisibleScopeEvents(otherChannelViewer, "wsp_1", "mem_other_channel", "chn_beta"))

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_alpha", storeim.StreamEvent{
		EventSeq: 12,
		EventID:  "evt_persisted",
		Type:     "runtime.task_created",
		Payload:  map[string]any{"taskId": "tsk_1"},
	})

	event := receiveHubEvent(t, viewer)
	require.Equal(t, "runtime.task_created", event.Type)
	require.Equal(t, "chn_alpha", event.ScopeID)
	require.Equal(t, "wsp_1", event.WorkspaceID)
	require.Equal(t, "mem_viewer", event.MemberID)
	require.Equal(t, int64(0), event.EventSeq)
	require.Empty(t, event.EventID)
	require.Equal(t, map[string]any{"taskId": "tsk_1"}, event.Payload)

	requireNoHubEvent(t, otherChannelViewer)
	requireNoHubEvent(t, workspacePeer)
	requireNoHubEvent(t, outsideWorkspace)
}

func TestStreamEventHubSubscribeScopeForUserAddsScopeToExistingConnections(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	target := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_target", "usr_target")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	defer target.Close()
	defer peer.Close()

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_alpha", storeim.StreamEvent{
		Type:    "runtime.task_created",
		Payload: map[string]any{"taskId": "tsk_pre"},
	})
	requireNoHubEvent(t, target)

	hub.SubscribeScopeForUser("wsp_1", "usr_target", "chn_alpha")

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_alpha", storeim.StreamEvent{
		Type:    "runtime.task_created",
		Payload: map[string]any{"taskId": "tsk_post"},
	})

	event := receiveHubEvent(t, target)
	require.Equal(t, "runtime.task_created", event.Type)
	require.Equal(t, "chn_alpha", event.ScopeID)
	require.Equal(t, "mem_target", event.MemberID)
	require.Equal(t, map[string]any{"taskId": "tsk_post"}, event.Payload)
	requireNoHubEvent(t, peer)
}

func TestStreamEventHubSubscribeScopeForUserAcrossMultipleConnections(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	desktop := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_desktop", "usr_target")
	mobile := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_mobile", "usr_target")
	defer desktop.Close()
	defer mobile.Close()

	hub.SubscribeScopeForUser("wsp_1", "usr_target", "chn_alpha")

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_alpha", storeim.StreamEvent{
		Type: "runtime.agent_wake_status",
	})

	for _, sub := range []*StreamEventSubscription{desktop, mobile} {
		event := receiveHubEvent(t, sub)
		require.Equal(t, "runtime.agent_wake_status", event.Type)
		require.Equal(t, "chn_alpha", event.ScopeID)
	}
}

func TestStreamEventHubUnsubscribeScopeForUserRemovesScopeFromExistingConnections(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	target := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_target", "usr_target")
	peer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_peer", "usr_peer")
	defer target.Close()
	defer peer.Close()

	require.True(t, hub.SubscribeVisibleScopeEvents(target, "wsp_1", "mem_target", "chn_alpha"))
	require.True(t, hub.SubscribeVisibleScopeEvents(peer, "wsp_1", "mem_peer", "chn_alpha"))

	hub.UnsubscribeScopeForUser("wsp_1", "usr_target", "chn_alpha")

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_alpha", storeim.StreamEvent{
		Type:    "runtime.task_updated",
		Payload: map[string]any{"taskId": "tsk_after"},
	})

	requireNoHubEvent(t, target)
	event := receiveHubEvent(t, peer)
	require.Equal(t, "runtime.task_updated", event.Type)
	require.Equal(t, "chn_alpha", event.ScopeID)
}

func TestStreamEventHubUnsubscribeScopeForUserAcrossMultipleConnections(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	desktop := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_desktop", "usr_target")
	mobile := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_mobile", "usr_target")
	defer desktop.Close()
	defer mobile.Close()

	require.True(t, hub.SubscribeVisibleScopeEvents(desktop, "wsp_1", "mem_desktop", "chn_alpha"))
	require.True(t, hub.SubscribeVisibleScopeEvents(mobile, "wsp_1", "mem_mobile", "chn_alpha"))

	hub.UnsubscribeScopeForUser("wsp_1", "usr_target", "chn_alpha")

	hub.PublishScopeLiveOnlyEvent("wsp_1", "chn_alpha", storeim.StreamEvent{
		Type: "runtime.agent_wake_status",
	})

	requireNoHubEvent(t, desktop)
	requireNoHubEvent(t, mobile)
}

func TestStreamEventHubScopeLiveOnlyWithoutWorkspaceOrScopeIsDropped(t *testing.T) {
	hub := NewStreamEventHub()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	viewer := hub.SubscribeStreamEvents(ctx, "wsp_1", "mem_viewer", "usr_viewer")
	defer viewer.Close()
	require.True(t, hub.SubscribeVisibleScopeEvents(viewer, "wsp_1", "mem_viewer", "chn_alpha"))

	hub.PublishScopeLiveOnlyEvent("", "chn_alpha", storeim.StreamEvent{Type: "runtime.task_created"})
	hub.PublishScopeLiveOnlyEvent("wsp_1", "", storeim.StreamEvent{Type: "runtime.task_created"})
	hub.PublishScopeLiveOnlyEvent("   ", "  ", storeim.StreamEvent{Type: "runtime.task_created"})

	requireNoHubEvent(t, viewer)
}

type recordingStreamEventPublisher struct {
	events        []storeim.StreamEvent
	visibleEvents []storeim.StreamEvent
}

func (publisher *recordingStreamEventPublisher) PublishStreamEvents(events []storeim.StreamEvent) {
	publisher.events = append(publisher.events, events...)
}

func (publisher *recordingStreamEventPublisher) PublishVisibleScopeEvents(events []storeim.StreamEvent) {
	publisher.visibleEvents = append(publisher.visibleEvents, events...)
}

type fakeStreamEventWriter struct {
	scopeEvents      []storeim.StreamEvent
	scopeAppendCalls int
}

func (writer *fakeStreamEventWriter) AppendScopeEvent(context.Context, storeim.ScopeEventRequest) ([]storeim.StreamEvent, error) {
	writer.scopeAppendCalls++
	return writer.scopeEvents, nil
}

func (writer *fakeStreamEventWriter) AppendMemberEvent(context.Context, string, string, string, storeim.StreamEvent, *storeim.ConversationIndexItem) (storeim.StreamEvent, error) {
	return storeim.StreamEvent{WorkspaceID: "wsp_1", MemberID: "mem_1", EventSeq: 1, Type: "conversation.updated"}, nil
}

func (writer *fakeStreamEventWriter) AppendMemberWorkspaceEvent(context.Context, string, string, storeim.StreamEvent) (storeim.StreamEvent, error) {
	return storeim.StreamEvent{WorkspaceID: "wsp_1", MemberID: "mem_1", EventSeq: 1, Type: "conversation.bulk.read"}, nil
}

func (writer *fakeStreamEventWriter) AppendConversationBulkReadEvent(context.Context, string, string, storeim.StreamEvent, []string) (storeim.StreamEvent, error) {
	return storeim.StreamEvent{WorkspaceID: "wsp_1", MemberID: "mem_1", EventSeq: 1, Type: "conversation.bulk.read"}, nil
}

func (writer *fakeStreamEventWriter) ArchiveIndexedConversation(_ context.Context, request storeim.IndexedConversationArchiveRequest) (storeim.StreamEvent, storeim.ConversationIndexItem, error) {
	return storeim.StreamEvent{WorkspaceID: request.WorkspaceID, MemberID: request.MemberID, ConversationID: request.ConversationID, EventSeq: 1, Type: request.Event.Type}, storeim.ConversationIndexItem{WorkspaceID: request.WorkspaceID, MemberID: request.MemberID, ConversationID: request.ConversationID}, nil
}