package im

import (
	"context"
	"database/sql"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api"
	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/dto/converter"
	"github.com/tidbcloud/AgentTeam/internal/runtimecontrol"
	messageservice "github.com/tidbcloud/AgentTeam/internal/service/message"
	"github.com/tidbcloud/AgentTeam/internal/store/identity"
	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/rewind"
	"github.com/tidbcloud/AgentTeam/internal/store/workspace"
	unreadservice "github.com/tidbcloud/AgentTeam/internal/unread"
)

type recordingStreamWriter struct {
	scopeRequests     []storeim.ScopeEventRequest
	memberRequests    []recordedMemberEvent
	workspaceRequests []recordedWorkspaceEvent
	archiveRequests   []storeim.IndexedConversationArchiveRequest
	archived          storeim.ConversationIndexItem
	scopeErr          error
	memberErr         error
	archiveErr        error
}

type recordedMemberEvent struct {
	workspaceID  string
	memberID     string
	scopeID      string
	event        storeim.StreamEvent
	conversation *storeim.ConversationIndexItem
}

type recordedWorkspaceEvent struct {
	workspaceID string
	memberID    string
	event       storeim.StreamEvent
}

func (writer *recordingStreamWriter) AppendScopeEvent(_ context.Context, request storeim.ScopeEventRequest) ([]storeim.StreamEvent, error) {
	writer.scopeRequests = append(writer.scopeRequests, request)
	if writer.scopeErr != nil {
		return nil, writer.scopeErr
	}
	return []storeim.StreamEvent{{WorkspaceID: request.WorkspaceID, MemberID: "mem_1", EventSeq: 1, Type: request.Event.Type}}, nil
}

func (writer *recordingStreamWriter) AppendMemberEvent(_ context.Context, workspaceID string, memberID string, scopeID string, event storeim.StreamEvent, conversation *storeim.ConversationIndexItem) (storeim.StreamEvent, error) {
	writer.memberRequests = append(writer.memberRequests, recordedMemberEvent{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID, event: event, conversation: conversation})
	if writer.memberErr != nil {
		return storeim.StreamEvent{}, writer.memberErr
	}
	return storeim.StreamEvent{WorkspaceID: workspaceID, MemberID: memberID, EventSeq: 1, Type: event.Type}, nil
}

func (writer *recordingStreamWriter) AppendMemberWorkspaceEvent(_ context.Context, workspaceID string, memberID string, event storeim.StreamEvent) (storeim.StreamEvent, error) {
	writer.workspaceRequests = append(writer.workspaceRequests, recordedWorkspaceEvent{workspaceID: workspaceID, memberID: memberID, event: event})
	return storeim.StreamEvent{WorkspaceID: workspaceID, MemberID: memberID, EventSeq: 1, Type: event.Type}, nil
}

func (writer *recordingStreamWriter) AppendConversationBulkReadEvent(_ context.Context, workspaceID string, memberID string, event storeim.StreamEvent, _ []string) (storeim.StreamEvent, error) {
	writer.workspaceRequests = append(writer.workspaceRequests, recordedWorkspaceEvent{workspaceID: workspaceID, memberID: memberID, event: event})
	return storeim.StreamEvent{WorkspaceID: workspaceID, MemberID: memberID, EventSeq: 1, Type: event.Type}, nil
}

func (writer *recordingStreamWriter) ArchiveIndexedConversation(_ context.Context, request storeim.IndexedConversationArchiveRequest) (storeim.StreamEvent, storeim.ConversationIndexItem, error) {
	writer.archiveRequests = append(writer.archiveRequests, request)
	if writer.archiveErr != nil {
		return storeim.StreamEvent{}, storeim.ConversationIndexItem{}, writer.archiveErr
	}
	return storeim.StreamEvent{WorkspaceID: request.WorkspaceID, MemberID: request.MemberID, ConversationID: request.ConversationID, EventSeq: 1, Type: request.Event.Type}, writer.archived, nil
}

func TestPublishMessagePinInvalidationsForRewindGroupsRemovedMessagesByScope(t *testing.T) {
	writer := &recordingStreamWriter{}
	plan := rewind.Plan{
		ScopeID:          "chn_1",
		BroadcastScopeID: "chn_1",
		SuffixMessageIDs: []string{"msg_1", " ", "msg_2"},
		CascadeReplyThreads: []rewind.CascadeReplyThread{{
			ScopeID:    "thr_1",
			MessageIDs: []string{"msg_3"},
		}, {
			ScopeID:    "",
			MessageIDs: []string{"msg_ignored"},
		}},
	}

	err := publishMessagePinInvalidationsForRewind(context.Background(), writer, "wsp_1", plan)
	require.NoError(t, err)
	require.Len(t, writer.scopeRequests, 2)

	requestsByScope := make(map[string]storeim.ScopeEventRequest, len(writer.scopeRequests))
	for _, request := range writer.scopeRequests {
		requestsByScope[request.ScopeID] = request
	}
	require.Equal(t, "wsp_1", requestsByScope["chn_1"].WorkspaceID)
	require.Equal(t, imEventTypeMessagePinsInvalidated, requestsByScope["chn_1"].Event.Type)
	require.Equal(t, "chn_1", requestsByScope["chn_1"].Event.EntityID)
	require.Equal(t, []string{"msg_1", "msg_2"}, requestsByScope["chn_1"].Event.Payload["messageIds"])
	require.Equal(t, "wsp_1", requestsByScope["thr_1"].WorkspaceID)
	require.Equal(t, imEventTypeMessagePinsInvalidated, requestsByScope["thr_1"].Event.Type)
	require.Equal(t, "thr_1", requestsByScope["thr_1"].Event.EntityID)
	require.Equal(t, []string{"msg_3"}, requestsByScope["thr_1"].Event.Payload["messageIds"])
}

func TestPublishMessagePinInvalidationsForReplyThreadRewindStaysInThreadScope(t *testing.T) {
	writer := &recordingStreamWriter{}
	plan := rewind.Plan{
		ScopeID:             "thr_1",
		BroadcastScopeID:    "chn_1",
		ReplyThreadScopeID:  "thr_1",
		SuffixMessageIDs:    []string{"msg_thread_1"},
		CascadeReplyThreads: nil,
	}

	err := publishMessagePinInvalidationsForRewind(context.Background(), writer, "wsp_1", plan)
	require.NoError(t, err)
	require.Len(t, writer.scopeRequests, 1)
	require.Equal(t, "thr_1", writer.scopeRequests[0].ScopeID)
	require.Equal(t, []string{"msg_thread_1"}, writer.scopeRequests[0].Event.Payload["messageIds"])
}

func TestPublishMessageUpdatedEventCarriesSharedPatchOnly(t *testing.T) {
	writer := &recordingStreamWriter{}
	editedAt := time.Date(2026, 6, 19, 10, 5, 0, 0, time.UTC)
	appMessage := message.AppMessage{
		ID:               "msg_1",
		Seq:              42,
		ChannelID:        "chn_1",
		Content:          "edited content",
		Metadata:         map[string]any{"mentions": []any{map[string]any{"type": "agent", "id": "agt_1", "token": "helper"}}},
		EditedAt:         &editedAt,
		EditedByMemberID: "mem_1",
		EditRevision:     2,
		PinState:         &message.MessagePinState{},
	}

	err := publishMessageUpdatedEvent(context.Background(), writer, "wsp_1", "chn_1", appMessage)

	require.NoError(t, err)
	require.Len(t, writer.scopeRequests, 1)
	request := writer.scopeRequests[0]
	require.Equal(t, "wsp_1", request.WorkspaceID)
	require.Equal(t, "chn_1", request.ScopeID)
	require.Nil(t, request.Conversation)
	require.Equal(t, imEventTypeMessageUpdated, request.Event.Type)
	require.Equal(t, "msg_1", request.Event.EntityID)
	require.NotNil(t, request.Event.EntitySeq)
	require.EqualValues(t, 42, *request.Event.EntitySeq)
	require.Equal(t, editedAt, request.Event.CreatedAt)
	require.Equal(t, "chn_1", request.Event.Payload["scopeId"])
	require.Equal(t, "msg_1", request.Event.Payload["messageId"])
	require.Equal(t, "edited content", request.Event.Payload["content"])
	require.Equal(t, "mem_1", request.Event.Payload["editedByMemberId"])
	require.EqualValues(t, 2, request.Event.Payload["editRevision"])
	require.NotContains(t, request.Event.Payload, "message")
	require.NotContains(t, request.Event.Payload, "pinState")
	require.NotContains(t, request.Event.Payload, "reactions")
}

type fakeHumanWriter struct {
	message  message.AppMessage
	delivery message.MessageDelivery
}

func (writer fakeHumanWriter) ValidateHumanMessage(context.Context, identity.User, string, string, string, messageservice.SendOptions) error {
	return nil
}

func (writer fakeHumanWriter) SendHumanMessage(context.Context, identity.User, string, string, string, []string, messageservice.SendOptions) (message.AppMessage, message.MessageDelivery, error) {
	return writer.message, writer.delivery, nil
}

type recordingAgentNotifier struct {
	recipients []message.AgentRecipient
	messages   []message.ReceiveMessage
	err        error
}

func (notifier *recordingAgentNotifier) NotifyAgentMessage(_ context.Context, recipient message.AgentRecipient, _ int64, message message.ReceiveMessage) error {
	notifier.recipients = append(notifier.recipients, recipient)
	notifier.messages = append(notifier.messages, message)
	return notifier.err
}

type recordingScopeFanoutMarker struct {
	ids []string
}

func (marker *recordingScopeFanoutMarker) MarkScopeFanoutDeliveredForMessage(_ context.Context, messageID string) error {
	marker.ids = append(marker.ids, messageID)
	return nil
}

type fakeAgentDeliveryStore struct {
	message  message.AppMessage
	delivery message.MessageDelivery
	statuses []string
}

func (store *fakeAgentDeliveryStore) PrepareMessageAgentDeliveryForUser(context.Context, string, string, string, string) (message.AppMessage, message.MessageDelivery, error) {
	return store.message, store.delivery, nil
}

func (store *fakeAgentDeliveryStore) CommitMessageAgentDeliveryForUser(context.Context, string, string, string, string) (message.AppMessage, error) {
	store.statuses = append(store.statuses, "committed")
	return store.message, nil
}

func (store *fakeAgentDeliveryStore) MarkMessageAgentDeliveryStatusForUser(_ context.Context, _ string, _ string, _ string, _ string, status string) (message.AppMessage, error) {
	store.statuses = append(store.statuses, status)
	return store.message, nil
}

func TestMessagesCreateAppendsScopeStreamEvent(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	createdAt := time.Date(2026, 6, 10, 10, 0, 0, 0, time.UTC)
	handler := &MessagesHandler{
		streamWriter: streamWriter,
		writer: fakeHumanWriter{message: message.AppMessage{
			ID:        "msg_1",
			Seq:       42,
			ChannelID: "chn_1",
			Content:   "hello",
			CreatedAt: createdAt,
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				PrincipalID: "usr_1",
				MemberID:    "mem_1",
				DisplayName: "Owner",
			},
		}, delivery: message.MessageDelivery{MessageID: "msg_1", Seq: 42}},
	}

	response, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "hello"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.scopeRequests, 1)
	request := streamWriter.scopeRequests[0]
	require.Equal(t, "wsp_1", request.WorkspaceID)
	require.Equal(t, "chn_1", request.ScopeID)
	require.Equal(t, imEventTypeMessageCreated, request.Event.Type)
	require.Equal(t, "msg_1", request.Event.EntityID)
	require.NotNil(t, request.Event.EntitySeq)
	require.Equal(t, int64(42), *request.Event.EntitySeq)
	require.NotNil(t, request.Conversation)
	require.Equal(t, "chn_1", request.Conversation.ScopeID)
	message, ok := request.Event.Payload["message"].(dto.AppMessage)
	require.True(t, ok)
	require.Equal(t, "msg_1", message.ID)
	require.Equal(t, "hello", message.Content)
}

func TestPublishServiceAccountArticleOmitsPrivatePublisherFromSharedEvent(t *testing.T) {
	writer := &recordingStreamWriter{}
	appMessage := message.AppMessage{
		ID:          "msg_article",
		Seq:         7,
		ChannelID:   "scope_service_account",
		ChannelType: "service_account",
		MessageType: message.MessageTypeServiceAccountPost,
		Content:     "Article body",
		CreatedAt:   time.Date(2026, 7, 13, 10, 0, 0, 0, time.UTC),
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeService,
			PrincipalID: "sacct_guide",
			MemberID:    "mem_service_account",
			DisplayName: "AgentTeam Guide",
			ServiceKind: message.AuthorServiceKindServiceAccount,
		},
		Metadata: map[string]any{
			"serviceAccountPost": map[string]any{"title": "Notification guide"},
			"serviceAccountPublisher": map[string]any{
				"bylineActorRef": map[string]any{"memberId": "mem_private"},
			},
		},
	}

	err := publishMessageCreatedEvent(
		context.Background(),
		writer,
		"wsp_1",
		appMessage.ChannelID,
		appMessage,
		message.MessageDelivery{},
	)

	require.NoError(t, err)
	require.Len(t, writer.scopeRequests, 1)
	payloadMessage, ok := writer.scopeRequests[0].Event.Payload["message"].(dto.AppMessage)
	require.True(t, ok)
	require.NotContains(t, payloadMessage.Metadata, "serviceAccountPublisher")
	require.Equal(t, map[string]any{"title": "Notification guide"}, payloadMessage.Metadata["serviceAccountPost"])
}

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

	createdAt := time.Date(2026, 7, 11, 9, 30, 0, 0, time.UTC)
	item := conversationIndexFromMessage(dto.AppMessage{
		ID:          "msg_service_account",
		Seq:         4,
		ChannelID:   "sasc_news",
		ChannelType: "service_account",
		AuthorProfile: &dto.AuthorProfile{
			ActorType:   "service",
			PrincipalID: "sacct_news",
			MemberID:    "mem_service_news",
			DisplayName: "Workspace News",
			AvatarURL:   "https://example.com/news.png",
			ServiceKind: "service_account",
			ServiceSlug: "workspace-news",
		},
		Content:   "A service-account announcement",
		CreatedAt: createdAt,
	})

	require.NotNil(t, item)
	require.Equal(t, "serviceAccount:sasc_news", item.ConversationID)
	require.Equal(t, "sasc_news", item.ScopeID)
	require.Equal(t, message.ConversationKindServiceAccount, item.Kind)
	require.Equal(t, createdAt, item.SortAt)
	require.Equal(t, "Workspace News", item.Projection["title"])
	require.Equal(t, "workspace-news", item.Projection["subtitle"])
	require.Equal(t, "service", item.Projection["peerType"])
	require.Equal(t, "sacct_news", item.Projection["peerId"])
	require.Equal(t, "/inbox/service-accounts/sacct_news", item.Projection["route"])
	require.Equal(t, "A service-account announcement", item.Projection["preview"])
	require.Equal(t, "active", item.Projection["subscriptionState"])
	require.Equal(t, createdAt, item.Projection["lastMessageAt"])
	require.IsType(t, dto.AppMessage{}, item.Projection["lastMessage"])
}

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

	item := conversationIndexFromMessage(dto.AppMessage{
		ID:          "msg_task",
		Seq:         9,
		ChannelID:   "tsk_task_index",
		ChannelType: message.ConversationKindTask,
		Content:     "task body",
		CreatedAt:   time.Date(2026, 7, 15, 12, 0, 0, 0, time.UTC),
	})

	require.Nil(t, item, "task messages must not materialize an Inbox conversation index row")
}

func TestMessagesCreateRestoresAgentFanoutAndMarksOutboxAfterStream(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	notifier := &recordingAgentNotifier{}
	marker := &recordingScopeFanoutMarker{}
	handler := &MessagesHandler{
		streamWriter:      streamWriter,
		agentNotifier:     notifier,
		scopeFanoutMarker: marker,
		writer: fakeHumanWriter{message: message.AppMessage{
			ID:        "msg_1",
			Seq:       42,
			ChannelID: "chn_1",
			Content:   "hello agent",
			CreatedAt: time.Date(2026, 6, 10, 10, 5, 0, 0, time.UTC),
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				PrincipalID: "usr_1",
				MemberID:    "mem_1",
				DisplayName: "Owner",
			},
		}, delivery: message.MessageDelivery{
			MessageID: "msg_1",
			Seq:       42,
			Message: message.ReceiveMessage{
				MessageID:   "msg_1",
				TargetKey:   "chn_1",
				ChannelType: "channel",
				Content:     "hello agent",
			},
			WakeRecipients: []message.AgentRecipient{{AgentID: "agt_1", AgentMemberID: "mem_agent_1"}},
		}},
	}

	response, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "hello agent"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, notifier.messages, 1)
	require.Equal(t, "msg_1", notifier.messages[0].MessageID)
	require.Equal(t, "chn_1", notifier.messages[0].TargetKey)
	require.Equal(t, int64(42), notifier.messages[0].TargetSeq)
	require.Len(t, streamWriter.scopeRequests, 1)
	require.Equal(t, imEventTypeMessageCreated, streamWriter.scopeRequests[0].Event.Type)
	require.Equal(t, []string{"msg_1"}, marker.ids)
}

func TestNotifyAgentsStrictAttachesRecipientTeamContexts(t *testing.T) {
	notifier := &recordingAgentNotifier{}
	handler := &MessagesHandler{agentNotifier: notifier}

	err := handler.notifyAgentsStrict(context.Background(), "wsp_1", message.MessageDelivery{
		MessageID: "msg_team",
		Seq:       77,
		Message: message.ReceiveMessage{
			MessageID:   "msg_team",
			TargetKey:   "chn_1",
			ChannelType: "channel",
			Content:     "@core-team please coordinate",
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:       "agt_core",
			AgentMemberID: "mem_agent",
			TeamContexts: []message.TeamDeliveryContext{{
				TeamID:                    "team_core",
				TeamHandle:                "core-team",
				TeamDisplayName:           "Core Team",
				TeamDescription:           "Core ownership",
				TeamMemberID:              "wtm_agent",
				RoleLabel:                 "Builder",
				RoleOverlayPromptSnapshot: "Focus on delivery.",
				RoleOverlayVersion:        3,
				MemberRoster: []message.TeamDeliveryRosterMember{{
					ActorKind:         "agent",
					ActorID:           "mem_agent",
					DisplayName:       "Core Agent",
					Handle:            "core-agent",
					RoleLabel:         "Builder",
					RoleOverlayPrompt: "Focus on delivery.",
					WakeOnTeamMention: true,
					DeliveryKind:      "wake",
					Self:              true,
				}},
			}},
		}},
	})

	require.NoError(t, err)
	require.Len(t, notifier.messages, 1)
	contexts := notifier.messages[0].TeamContexts
	require.Len(t, contexts, 1)
	require.Equal(t, "team_core", contexts[0].TeamID)
	require.Equal(t, "Focus on delivery.", contexts[0].RoleOverlayPrompt)
	require.Len(t, contexts[0].MemberRoster, 1)
	require.True(t, contexts[0].MemberRoster[0].Self)
	require.Equal(t, "Focus on delivery.", contexts[0].MemberRoster[0].RoleOverlayPrompt)
	require.True(t, contexts[0].MemberRoster[0].WakeOnTeamMention)
}

func TestMessagesCreateReturnsCommittedMessageWhenAgentFanoutFails(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	notifier := &recordingAgentNotifier{err: errors.New("machine offline")}
	marker := &recordingScopeFanoutMarker{}
	handler := &MessagesHandler{
		streamWriter:      streamWriter,
		agentNotifier:     notifier,
		scopeFanoutMarker: marker,
		writer: fakeHumanWriter{message: message.AppMessage{
			ID:        "msg_failed",
			Seq:       7,
			ChannelID: "chn_1",
			Content:   "hello",
			CreatedAt: time.Now().UTC(),
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				PrincipalID: "usr_1",
				MemberID:    "mem_1",
				DisplayName: "Owner",
			},
		}, delivery: message.MessageDelivery{
			MessageID:      "msg_failed",
			Seq:            7,
			WakeRecipients: []message.AgentRecipient{{AgentID: "agt_1"}},
		}},
	}

	response, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "hello"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	message := response.Body.(dto.AppMessage)
	require.Equal(t, "msg_failed", message.ID)
	require.Len(t, streamWriter.scopeRequests, 1)
	require.Equal(t, imEventTypeMessageCreated, streamWriter.scopeRequests[0].Event.Type)
	require.Len(t, notifier.messages, 1)
	require.Empty(t, marker.ids)
}

func TestMessagesCreateReturnsCommittedMessageWhenStreamFanoutFailsBeforeNotify(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{scopeErr: errors.New("stream unavailable")}
	notifier := &recordingAgentNotifier{}
	marker := &recordingScopeFanoutMarker{}
	handler := &MessagesHandler{
		streamWriter:      streamWriter,
		agentNotifier:     notifier,
		scopeFanoutMarker: marker,
		writer: fakeHumanWriter{message: message.AppMessage{
			ID:        "msg_stream_failed",
			Seq:       8,
			ChannelID: "chn_1",
			Content:   "hello",
			CreatedAt: time.Now().UTC(),
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				PrincipalID: "usr_1",
				MemberID:    "mem_1",
				DisplayName: "Owner",
			},
		}, delivery: message.MessageDelivery{
			MessageID: "msg_stream_failed",
			Seq:       8,
			Message: message.ReceiveMessage{
				MessageID: "msg_stream_failed",
				TargetKey: "chn_1",
				Content:   "hello",
			},
			WakeRecipients: []message.AgentRecipient{{AgentID: "agt_1"}},
		}},
	}

	response, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "hello"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	message := response.Body.(dto.AppMessage)
	require.Equal(t, "msg_stream_failed", message.ID)
	require.Len(t, streamWriter.scopeRequests, 1)
	require.Empty(t, notifier.messages)
	require.Empty(t, marker.ids)
}

func TestSubmitAgentDeliveryPublishesFailureStatusUpdate(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	store := &fakeAgentDeliveryStore{
		message: message.AppMessage{
			ID:        "msg_1",
			ChannelID: "chn_1",
			Content:   "hello agent",
			CreatedAt: time.Now().UTC(),
		},
		delivery: message.MessageDelivery{
			MessageID: "msg_1",
			Seq:       42,
			Message: message.ReceiveMessage{
				MessageID: "msg_1",
				TargetKey: "chn_1",
				Content:   "hello agent",
			},
			WakeRecipients: []message.AgentRecipient{{AgentID: "agt_1"}},
		},
	}
	handler := &MessagesHandler{
		agentDeliveries: store,
		agentNotifier:   &recordingAgentNotifier{err: errors.New("machine offline")},
		streamWriter:    streamWriter,
	}

	_, err := handler.SubmitAgentDelivery(imWorkspaceJSON(
		dto.AgentDeliveryCreateOptions{InterruptPolicy: message.AgentInterruptPolicyCurrentRun},
		gin.Param{Key: "messageId", Value: "msg_1"},
	))
	require.Error(t, err)
	require.Equal(t, []string{message.AgentDeliveryStatusPending, message.AgentDeliveryStatusFailed}, store.statuses)
	require.Len(t, streamWriter.scopeRequests, 2)
	for _, request := range streamWriter.scopeRequests {
		require.Equal(t, "wsp_1", request.WorkspaceID)
		require.Equal(t, "chn_1", request.ScopeID)
		require.Equal(t, imEventTypeMessageDelivery, request.Event.Type)
		require.Equal(t, "msg_1", request.Event.EntityID)
		require.Equal(t, "chn_1", request.Event.Payload["channelId"])
		require.Equal(t, []string{"msg_1"}, request.Event.Payload["messageIds"])
	}
}

func TestMessagesCreatePublishesMentionFanoutMetadataInMessageEvent(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	handler := &MessagesHandler{
		streamWriter: streamWriter,
		writer: fakeHumanWriter{message: message.AppMessage{
			ID:        "msg_mention",
			Seq:       3,
			ChannelID: "chn_1",
			Content:   "hello @teammate",
			CreatedAt: time.Now().UTC(),
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				PrincipalID: "usr_author",
				MemberID:    "mem_author",
				DisplayName: "Author",
			},
		}, delivery: message.MessageDelivery{
			MessageID: "msg_mention",
			Seq:       3,
			MentionContext: message.MentionContext{
				ChannelType: "channel",
			},
			MentionedHumans: []message.HumanMentionRecipient{{UserID: "usr_mentioned", MemberID: "mem_mentioned"}},
		}},
	}

	_, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "hello @teammate"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Len(t, streamWriter.scopeRequests, 1)
	payload := streamWriter.scopeRequests[0].Event.Payload
	mention, ok := payload["mention"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "msg_mention", mention["messageId"])
	require.Equal(t, "channel", mention["channelType"])
	recipients, ok := payload["mentionRecipients"].([]map[string]string)
	require.True(t, ok)
	require.Equal(t, []map[string]string{{"userId": "usr_mentioned", "memberId": "mem_mentioned"}}, recipients)
}

func TestTaskMessageCreatedProjectionCarriesNotificationRouteContext(t *testing.T) {
	writer := &recordingStreamWriter{}
	appMessage := message.AppMessage{
		ID:          "msg_task_mention",
		Seq:         8,
		ChannelID:   "tsk_scope",
		ChannelType: message.ConversationKindTask,
		Content:     "@mem_reader please review",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			PrincipalID: "usr_author",
			MemberID:    "mem_author",
			DisplayName: "Author",
		},
		TaskContext: &message.AppMessageTaskContext{
			ChannelID:  "chn_board",
			TaskNumber: 42,
		},
	}
	delivery := message.MessageDelivery{
		MessageID: "msg_task_mention",
		MentionedHumans: []message.HumanMentionRecipient{{
			UserID:   "usr_reader",
			MemberID: "mem_reader",
		}},
		MentionContext: message.MentionContext{
			ChannelType: "task",
		},
	}

	require.NoError(t, publishMessageCreatedEvent(t.Context(), writer, "wsp_1", "tsk_scope", appMessage, delivery))
	require.Len(t, writer.scopeRequests, 1)
	payload := writer.scopeRequests[0].Event.Payload
	projected, ok := payload["message"].(dto.AppMessage)
	require.True(t, ok)
	require.Equal(t, &dto.AppMessageTaskContext{ChannelID: "chn_board", TaskNumber: 42}, projected.TaskContext)
	mention, ok := payload["mention"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, map[string]any{"channelId": "chn_board", "taskNumber": int64(42)}, mention["taskContext"])
}

func TestTaskReactionNotificationProjectionCarriesRouteContext(t *testing.T) {
	notification := &message.HumanReactionNotification{
		ReactionID:  "rxn_task",
		MessageID:   "msg_task",
		ChannelID:   "tsk_scope",
		ChannelType: "task",
		TaskContext: &message.AppMessageTaskContext{
			ChannelID:  "chn_board",
			TaskNumber: 42,
		},
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			PrincipalID: "usr_reactor",
			MemberID:    "mem_reactor",
			DisplayName: "Reactor",
		},
		Emoji: "👍",
	}

	payload := reactionNotificationPayload(notification)
	require.Equal(t, map[string]any{"channelId": "chn_board", "taskNumber": int64(42)}, payload["taskContext"])
}

func TestMessagesCreatePublishesFollowupMessagesAndMarksTree(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	marker := &recordingScopeFanoutMarker{}
	now := time.Now().UTC()
	handler := &MessagesHandler{
		streamWriter:      streamWriter,
		scopeFanoutMarker: marker,
		writer: fakeHumanWriter{message: message.AppMessage{
			ID:        "msg_root",
			Seq:       10,
			ChannelID: "chn_1",
			Content:   "root",
			CreatedAt: now,
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				PrincipalID: "usr_1",
				MemberID:    "mem_1",
				DisplayName: "Owner",
			},
		}, delivery: message.MessageDelivery{
			MessageID: "msg_root",
			Seq:       10,
			FollowupDeliveries: []message.MessageDelivery{{
				MessageID: "msg_followup",
				Seq:       11,
				AppMessage: &message.AppMessage{
					ID:        "msg_followup",
					Seq:       11,
					ChannelID: "chn_1",
					Content:   "followup",
					CreatedAt: now.Add(time.Second),
					AuthorProfile: message.AuthorProfile{
						ActorType:   message.AuthorActorTypeService,
						ServiceKind: message.AuthorServiceKindApp,
						DisplayName: "AgentTeam",
					},
				},
			}},
		}},
	}

	_, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "root"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Len(t, streamWriter.scopeRequests, 2)
	require.Equal(t, "msg_root", streamWriter.scopeRequests[0].Event.EntityID)
	require.Equal(t, "msg_followup", streamWriter.scopeRequests[1].Event.EntityID)
	require.Equal(t, []string{"msg_root", "msg_followup"}, marker.ids)
}

func TestReceiveMessageForIMDeliveryDoesNotInventTargetSeqForStartedReplyThreadRoot(t *testing.T) {
	msg := receiveMessageForIMDelivery(message.MessageDelivery{
		MessageID: "msg_thread_root",
		Seq:       42,
		Message: message.ReceiveMessage{
			MessageID:   "msg_thread_root",
			TargetKey:   "thr_1",
			ChannelType: "thread",
			Content:     "@agent handle this in a thread",
		},
		StartedReplyThread: &message.ReplyThreadSummary{ThreadChannelID: "thr_1"},
	}, message.AgentRecipient{})

	require.Equal(t, "thr_1", msg.TargetKey)
	require.Zero(t, msg.TargetSeq)
}

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

	msg := receiveMessageForIMDelivery(message.MessageDelivery{
		MessageID: "tev_checkpoint",
		Seq:       99,
		Message: message.ReceiveMessage{
			MessageID:   "tev_checkpoint",
			TargetKey:   "tsk_scope",
			TargetSeq:   88,
			ChannelType: "task",
			MessageType: "task.checkpoint_added",
			Source: message.ContributionSource{
				RefType: message.ContributionSourceRefTypeTaskActivity,
				RefID:   "tev_checkpoint",
			},
		},
	}, message.AgentRecipient{})

	require.Zero(t, msg.TargetSeq)
}

func TestMessagesCreateDoesNotAppendStreamEventForIdempotentReplay(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	handler := &MessagesHandler{
		streamWriter: streamWriter,
		writer:       fakeHumanWriter{message: message.AppMessage{ID: "msg_1", Seq: 1, ChannelID: "chn_1", Content: "hello", CreatedAt: time.Now().UTC()}, delivery: message.MessageDelivery{MessageID: "msg_1", IdempotentReplay: true}},
	}

	_, err := handler.Create(imWorkspaceJSON(messageCreateRequest{Content: "hello"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Empty(t, streamWriter.scopeRequests)
}

type fakeConversationStore struct {
	item       message.ConversationItem
	page       message.ConversationPage
	bulkResult message.ConversationBulkActionResult
	actionErr  error
	loadErr    error
}

func (store fakeConversationStore) ListConversationsForUserFiltered(context.Context, string, string, int, message.ConversationCursor, message.ConversationKindFilter) (message.ConversationPage, error) {
	return store.page, nil
}
func (store fakeConversationStore) MarkConversationsReadForUser(context.Context, string, string, message.ConversationKindFilter) (message.ConversationBulkActionResult, error) {
	return store.bulkResult, nil
}
func (store fakeConversationStore) PerformConversationActionForUser(context.Context, string, string, string, string) (message.ConversationItem, error) {
	return store.item, store.actionErr
}
func (store fakeConversationStore) LoadConversationForUser(context.Context, string, string, string, bool) (message.ConversationItem, error) {
	return store.item, store.loadErr
}

type fakeConversationIndexStore struct {
	page        storeim.ConversationIndexPage
	listedKinds []string
	upserted    []storeim.ConversationIndexItem
}

func (store *fakeConversationIndexStore) ListConversations(_ context.Context, _ string, _ string, _ int, _ storeim.ConversationCursor, kinds []string) (storeim.ConversationIndexPage, error) {
	store.listedKinds = append([]string{}, kinds...)
	return store.page, nil
}
func (store *fakeConversationIndexStore) GetConversation(_ context.Context, _ string, _ string, _ string) (storeim.ConversationIndexItem, error) {
	if len(store.page.Items) == 0 {
		return storeim.ConversationIndexItem{}, nil
	}
	return store.page.Items[0], nil
}
func (store *fakeConversationIndexStore) LookupMemberVisibleConversationProjectionsByIDs(_ context.Context, _ string, _ string, conversationIDs []string) ([]storeim.ConversationIndexItem, error) {
	requested := make(map[string]struct{}, len(conversationIDs))
	for _, conversationID := range conversationIDs {
		requested[conversationID] = struct{}{}
	}
	items := make([]storeim.ConversationIndexItem, 0, len(conversationIDs))
	for _, item := range store.page.Items {
		if _, ok := requested[item.ConversationID]; ok {
			items = append(items, item)
		}
	}
	return items, nil
}
func (store *fakeConversationIndexStore) UpsertConversation(_ context.Context, _ string, _ string, item storeim.ConversationIndexItem) error {
	store.upserted = append(store.upserted, item)
	return nil
}

type fakeUnreadRuntime struct {
	states             map[string]unreadservice.ScopeUnreadState
	workspaceHasUnread bool
	scopeStateCalls    [][]string
	readEvents         []unreadservice.ReadCursorAdvanced
	activityReadEvents []unreadservice.ActivityReadCursorAdvanced
	manualEvents       []unreadservice.ManualUnreadChanged
	invalidations      []unreadservice.ScopeInvalidated
	invalidationDeltas []unreadservice.UnreadDelta
	refreshes          []fakeRosterRefresh
}

type fakeRosterRefresh struct {
	workspaceID string
	scopeID     string
	scopeType   unreadservice.ScopeType
	memberIDs   []string
}

func (runtime *fakeUnreadRuntime) ScopeStates(workspaceID string, memberID string, scopeIDs []string) map[string]unreadservice.ScopeUnreadState {
	if runtime == nil {
		return nil
	}
	runtime.scopeStateCalls = append(runtime.scopeStateCalls, append([]string(nil), scopeIDs...))
	results := make(map[string]unreadservice.ScopeUnreadState)
	if len(scopeIDs) == 0 {
		for scopeID, state := range runtime.states {
			results[scopeID] = state
		}
		return results
	}
	for _, scopeID := range scopeIDs {
		if state, ok := runtime.states[scopeID]; ok {
			state.WorkspaceID = workspaceID
			state.MemberID = memberID
			state.ScopeID = scopeID
			results[scopeID] = state
		}
	}
	return results
}

func (runtime *fakeUnreadRuntime) WorkspaceStates(workspaceID string, memberID string) map[string]unreadservice.ScopeUnreadState {
	return runtime.ScopeStates(workspaceID, memberID, nil)
}

func (runtime *fakeUnreadRuntime) WorkspaceHasUnread(string, string) bool {
	if runtime == nil {
		return false
	}
	if runtime.workspaceHasUnread {
		return true
	}
	for _, state := range runtime.states {
		if state.HasUnread() {
			return true
		}
	}
	return false
}

func (runtime *fakeUnreadRuntime) WorkspaceUnreadConversationCount(string, string) int64 {
	if runtime == nil {
		return 0
	}
	var count int64
	for _, state := range runtime.states {
		if state.HasUnread() {
			count++
		}
	}
	return count
}

func (runtime *fakeUnreadRuntime) ApplyMessage(unreadservice.MessageStored) (unreadservice.ApplyResult, error) {
	return unreadservice.ApplyResult{}, nil
}

func (runtime *fakeUnreadRuntime) ApplyNonMessageActivity(unreadservice.NonMessageActivityStored) (unreadservice.ApplyResult, error) {
	return unreadservice.ApplyResult{}, nil
}

func (runtime *fakeUnreadRuntime) ApplyMessageRewound(unreadservice.MessageRewound) (unreadservice.ApplyResult, error) {
	return unreadservice.ApplyResult{}, nil
}

func (runtime *fakeUnreadRuntime) ApplyReadCursor(event unreadservice.ReadCursorAdvanced) error {
	runtime.readEvents = append(runtime.readEvents, event)
	return nil
}

func (runtime *fakeUnreadRuntime) ApplyActivityReadCursor(event unreadservice.ActivityReadCursorAdvanced) error {
	runtime.activityReadEvents = append(runtime.activityReadEvents, event)
	return nil
}

func (runtime *fakeUnreadRuntime) ApplyManualUnread(event unreadservice.ManualUnreadChanged) error {
	runtime.manualEvents = append(runtime.manualEvents, event)
	return nil
}

func (runtime *fakeUnreadRuntime) ApplyScopeInvalidated(event unreadservice.ScopeInvalidated) error {
	runtime.invalidations = append(runtime.invalidations, event)
	return nil
}

func (runtime *fakeUnreadRuntime) ApplyScopeInvalidatedWithDeltas(event unreadservice.ScopeInvalidated) ([]unreadservice.UnreadDelta, error) {
	runtime.invalidations = append(runtime.invalidations, event)
	return append([]unreadservice.UnreadDelta(nil), runtime.invalidationDeltas...), nil
}

func (runtime *fakeUnreadRuntime) ApplyNotificationMuted(event unreadservice.ScopeInvalidated) error {
	runtime.invalidations = append(runtime.invalidations, event)
	return nil
}

func (runtime *fakeUnreadRuntime) RefreshScopeRoster(workspaceID string, scopeID string, scopeType unreadservice.ScopeType, memberIDs []string) error {
	runtime.refreshes = append(runtime.refreshes, fakeRosterRefresh{
		workspaceID: workspaceID,
		scopeID:     scopeID,
		scopeType:   scopeType,
		memberIDs:   append([]string(nil), memberIDs...),
	})
	return nil
}

type recordingUnreadSink struct {
	unreadUpdated []unreadservice.UnreadDelta
}

func (sink *recordingUnreadSink) EmitUnreadUpdated(_ string, delta unreadservice.UnreadDelta) {
	sink.unreadUpdated = append(sink.unreadUpdated, delta)
}

func TestConversationsListUsesIndexSourceAsConversationPage(t *testing.T) {
	gin.SetMode(gin.TestMode)
	index := &fakeConversationIndexStore{page: storeim.ConversationIndexPage{
		Items: []storeim.ConversationIndexItem{{
			ConversationID: "thread:thr_1",
			ScopeID:        "thr_1",
			Kind:           "thread",
			SortAt:         time.Date(2026, 6, 10, 11, 0, 0, 0, time.UTC),
			LastEventSeq:   8,
			Projection: map[string]any{
				"id":             "thread:thr_1",
				"kind":           "replyThread",
				"scopeId":        "thr_1",
				"route":          "/inbox/channel/chn_1/thread/thr_1",
				"title":          "Thread",
				"lastActivityAt": "2026-06-10T11:00:00Z",
				"lastMessageAt":  "2026-06-10T11:00:00Z",
				"unreadCount":    1,
			},
		}},
		HasMore: true,
		NextCursor: storeim.ConversationCursor{
			SortAt:         time.Date(2026, 6, 10, 11, 0, 0, 0, time.UTC),
			ConversationID: "thread:thr_1",
		},
	}}
	handler := &ConversationsHandler{
		conversationIndex: index,
		unreadRuntime: &fakeUnreadRuntime{states: map[string]unreadservice.ScopeUnreadState{
			"thr_1": {ScopeID: "thr_1", MessageUnreadCount: 2, ServingState: unreadservice.ServingStateReady},
		}},
	}
	request := imWorkspaceRequest()
	request.Gin.Request = httptest.NewRequest(http.MethodGet, "/?source=index&limit=10&kinds=replyThread", nil)

	response, err := handler.List(request)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Equal(t, []string{"thread"}, index.listedKinds)
	page := response.Body.(dto.ConversationPage)
	require.True(t, page.HasMore)
	require.NotEmpty(t, page.NextCursor)
	require.Len(t, page.Items, 1)
	require.Equal(t, "thread:thr_1", page.Items[0].ID)
	require.Equal(t, "replyThread", page.Items[0].Kind)
	require.Equal(t, int64(2), page.Items[0].UnreadCount)
}

func TestConversationsListKeepsIndexSourceEmptyWhenIndexIsEmpty(t *testing.T) {
	gin.SetMode(gin.TestMode)
	index := &fakeConversationIndexStore{}
	handler := &ConversationsHandler{
		conversationIndex: index,
		conversations:     fakeConversationStore{},
	}
	request := imWorkspaceRequest()
	request.Gin.Request = httptest.NewRequest(http.MethodGet, "/?source=index&kinds=channel", nil)

	response, err := handler.List(request)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Equal(t, []string{"channel"}, index.listedKinds)
	page := response.Body.(dto.ConversationPage)
	require.Empty(t, page.Items)
	require.False(t, page.HasMore)
	require.Empty(t, page.NextCursor)
}

func TestConversationsListOverlaysRuntimeUnread(t *testing.T) {
	gin.SetMode(gin.TestMode)
	unread := &fakeUnreadRuntime{states: map[string]unreadservice.ScopeUnreadState{
		"chn_1": {ScopeID: "chn_1", MessageUnreadCount: 2, ServingState: unreadservice.ServingStateReady},
		"tsk_1": {ScopeID: "tsk_1", ManualUnread: true, ServingState: unreadservice.ServingStateReady},
	}}
	handler := &ConversationsHandler{
		conversations: fakeConversationStore{page: message.ConversationPage{Items: []message.ConversationItem{
			{Kind: message.ConversationKindChannel, ScopeID: "chn_1", UnreadCount: 99, ManualUnread: true},
			{Kind: message.ConversationKindTask, ScopeID: "tsk_1", UnreadCount: 42, ManualUnread: false},
		}}},
		unreadRuntime: unread,
	}
	request := imWorkspaceRequest()
	request.Gin.Request = httptest.NewRequest(http.MethodGet, "/?limit=10", nil)

	response, err := handler.List(request)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	page := response.Body.(dto.ConversationPage)
	require.Len(t, page.Items, 2)
	require.Equal(t, int64(2), page.Items[0].UnreadCount)
	require.False(t, page.Items[0].ManualUnread)
	require.Equal(t, int64(0), page.Items[1].UnreadCount)
	require.True(t, page.Items[1].ManualUnread)
	require.Equal(t, [][]string{{"chn_1", "tsk_1"}}, unread.scopeStateCalls)
}

func TestConversationActionAppendsMemberStreamEventAndProjection(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	lastAt := time.Date(2026, 6, 10, 10, 30, 0, 0, time.UTC)
	handler := &ConversationsHandler{
		streamWriter: streamWriter,
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:           message.ConversationKindChannel,
			ScopeID:        "chn_1",
			Title:          "general",
			LastActivityAt: &lastAt,
			UnreadCount:    0,
		}},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionMarkRead}, gin.Param{Key: "conversationId", Value: "channel:chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, "wsp_1", recorded.workspaceID)
	require.Equal(t, "mem_1", recorded.memberID)
	require.Equal(t, "chn_1", recorded.scopeID)
	require.Equal(t, imEventTypeConversationRead, recorded.event.Type)
	require.NotNil(t, recorded.conversation)
	require.Equal(t, "channel:chn_1", recorded.conversation.ConversationID)
	require.Equal(t, "channel", recorded.conversation.Kind)
}

func TestConversationActionProjectionMarksUnpinAsAuthoritative(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	lastAt := time.Date(2026, 6, 10, 10, 30, 0, 0, time.UTC)
	handler := &ConversationsHandler{
		streamWriter: streamWriter,
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:           message.ConversationKindChannel,
			ScopeID:        "chn_1",
			Title:          "general",
			LastActivityAt: &lastAt,
			PinnedAt:       nil,
		}},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionUnpin}, gin.Param{Key: "conversationId", Value: "channel:chn_1"}))

	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.memberRequests, 1)
	require.NotNil(t, streamWriter.memberRequests[0].conversation)
	require.Nil(t, streamWriter.memberRequests[0].conversation.PinnedAt)
	require.True(t, streamWriter.memberRequests[0].conversation.PinnedAtSet, "unpin projections must explicitly clear the member-local pinned state")
}

func TestConversationDTOFromIndexProjectionUsesIndexPinnedAtAsAuthoritative(t *testing.T) {
	stalePinnedAt := time.Date(2026, 6, 10, 9, 0, 0, 0, time.UTC)
	item := storeim.ConversationIndexItem{
		ConversationID: "channel:chn_1",
		ScopeID:        "chn_1",
		Kind:           message.ConversationKindChannel,
		SortAt:         time.Date(2026, 6, 10, 10, 0, 0, 0, time.UTC),
		Projection: map[string]any{
			"id":       "channel:chn_1",
			"kind":     "channel",
			"scopeId":  "chn_1",
			"route":    "/inbox/channel/chn_1",
			"title":    "General",
			"pinnedAt": stalePinnedAt.Format(time.RFC3339Nano),
		},
	}

	dtoItem, ok := conversationDTOFromIndexProjection(item)

	require.True(t, ok)
	require.Nil(t, dtoItem.PinnedAt, "index row pinned_at must be the resolved pinned state, even if projection_json is stale")
}

func TestConversationDTOFromIndexProjectionSupportsServiceDM(t *testing.T) {
	item := storeim.ConversationIndexItem{
		ConversationID: "dm:dm_approval",
		ScopeID:        "dm_approval",
		Kind:           message.ConversationKindDM,
		SortAt:         time.Date(2026, 7, 11, 8, 35, 41, 0, time.UTC),
		Projection: map[string]any{
			"id":       "dm:dm_approval",
			"kind":     message.ConversationKindDM,
			"scopeId":  "dm_approval",
			"peerType": "service",
			"peerId":   "appinst_approval",
			"title":    "Approval",
		},
	}

	dtoItem, ok := conversationDTOFromIndexProjection(item)

	require.True(t, ok)
	require.Equal(t, "serviceDm", dtoItem.Kind)
	require.Equal(t, "/inbox/dm/dm_approval", dtoItem.Route)
}

func TestConversationDTOFromIndexProjectionBuildsThreadRouteFromCurrentScope(t *testing.T) {
	for _, testCase := range []struct {
		name       string
		scopeID    string
		projection map[string]any
		wantRoute  string
	}{
		{
			name:    "ordinary thread overwrites stale route",
			scopeID: "thr_1",
			projection: map[string]any{
				"id":              "thread:thr_1",
				"kind":            "replyThread",
				"scopeId":         "thr_1",
				"route":           "/inbox/invalid",
				"parentScopeId":   "chn_1",
				"parentScopeType": "channel",
				"title":           "Thread",
			},
			wantRoute: "/inbox/thread/thr_1",
		},
		{
			name:    "missing parent fact stays an inbox thread",
			scopeID: "thr_1",
			projection: map[string]any{
				"id":      "thread:thr_1",
				"kind":    "replyThread",
				"scopeId": "thr_1",
				"route":   "/inbox/invalid",
				"title":   "Thread",
			},
			wantRoute: "/inbox/thread/thr_1",
		},
		{
			name:    "knowledge thread overwrites stale direct route",
			scopeID: "thr / kb",
			projection: map[string]any{
				"id":              "thread:thr / kb",
				"kind":            "replyThread",
				"scopeId":         "stale-scope",
				"route":           "/kb/retrieval/thread/stale",
				"parentScopeType": "kb",
				"title":           "Knowledge thread",
			},
			wantRoute: "/inbox/kb/retrieval/thread/thr%20%2F%20kb",
		},
	} {
		t.Run(testCase.name, func(t *testing.T) {
			item := storeim.ConversationIndexItem{
				ConversationID: "thread:" + testCase.scopeID,
				ScopeID:        testCase.scopeID,
				Kind:           message.ConversationKindThread,
				SortAt:         time.Date(2026, 7, 17, 8, 0, 0, 0, time.UTC),
				Projection:     testCase.projection,
			}

			dtoItem, ok := conversationDTOFromIndexProjection(item)

			require.True(t, ok)
			require.Equal(t, testCase.wantRoute, dtoItem.Route)
			require.Equal(t, converter.InboxConversationRoute(message.ConversationItem{
				Kind:            message.ConversationKindThread,
				ScopeID:         testCase.scopeID,
				ParentScopeType: stringFromProjection(testCase.projection, "parentScopeType"),
			}), dtoItem.Route, "authoritative and index paths must route identical facts equivalently")
		})
	}
}

func TestConversationDTOFromIndexProjectionKeepsIndexKindAuthoritative(t *testing.T) {
	t.Run("task index kind overrides stale non-task projection", func(t *testing.T) {
		item := storeim.ConversationIndexItem{
			ConversationID: "task:tsk_leak",
			ScopeID:        "tsk_leak",
			Kind:           message.ConversationKindTask,
			SortAt:         time.Date(2026, 7, 15, 12, 0, 0, 0, time.UTC),
			Projection: map[string]any{
				"id":      "task:tsk_leak",
				"kind":    "channel",
				"scopeId": "tsk_leak",
				"route":   "/inbox/channel/tsk_leak",
				"title":   "Stale task row masking as channel",
				"task": map[string]any{
					"id":        "tsk_leak",
					"scopeId":   "tsk_leak",
					"channelId": "chn_1",
					"number":    7,
					"title":     "Canonical task",
					"status":    "todo",
				},
			},
		}

		dtoItem, ok := conversationDTOFromIndexProjection(item)

		require.True(t, ok)
		require.Equal(t, "task", dtoItem.Kind)
		require.Equal(t, "task:tsk_leak", dtoItem.ID)
		require.Equal(t, "tsk_leak", dtoItem.ScopeID)
		require.Equal(t, "/inbox/tasks/chn_1/7", dtoItem.Route)
	})

	t.Run("non-task index kind rejects stale task projection", func(t *testing.T) {
		item := storeim.ConversationIndexItem{
			ConversationID: "channel:chn_1",
			ScopeID:        "chn_1",
			Kind:           message.ConversationKindChannel,
			SortAt:         time.Date(2026, 7, 15, 12, 0, 0, 0, time.UTC),
			Projection: map[string]any{
				"id":      "task:tsk_leak",
				"kind":    "task",
				"scopeId": "tsk_leak",
				"route":   "/tasks/tsk_leak",
				"title":   "Legacy task payload",
			},
		}

		dtoItem, ok := conversationDTOFromIndexProjection(item)

		require.True(t, ok)
		require.Equal(t, "channel", dtoItem.Kind)
		require.Equal(t, "channel:chn_1", dtoItem.ID)
		require.Equal(t, "chn_1", dtoItem.ScopeID)
		require.Equal(t, "/inbox/channel/chn_1", dtoItem.Route)
	})
}

func TestConversationActionFeedsUnreadRuntime(t *testing.T) {
	gin.SetMode(gin.TestMode)
	unread := &fakeUnreadRuntime{states: map[string]unreadservice.ScopeUnreadState{
		"tsk_1": {ScopeID: "tsk_1", ServingState: unreadservice.ServingStateReady},
	}}
	sink := &recordingUnreadSink{}
	handler := &ConversationsHandler{
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:         message.ConversationKindTask,
			ScopeID:      "tsk_1",
			UnreadCount:  99,
			ManualUnread: true,
			LastReadSeq:  7,
		}},
		unreadRuntime: unread,
		unreadSink:    sink,
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionMarkRead}, gin.Param{Key: "conversationId", Value: "task:tsk_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	item := response.Body.(dto.ConversationItem)
	require.Equal(t, int64(0), item.UnreadCount)
	require.False(t, item.ManualUnread)
	require.Equal(t, []unreadservice.ReadCursorAdvanced{{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_1",
		ScopeID:     "tsk_1",
		LastReadSeq: 7,
	}}, unread.readEvents)
	require.Equal(t, []unreadservice.ActivityReadCursorAdvanced{{
		WorkspaceID: "wsp_1",
		MemberID:    "mem_1",
		ScopeID:     "tsk_1",
	}}, unread.activityReadEvents)
	require.Equal(t, []unreadservice.UnreadDelta{{
		WorkspaceID:        "wsp_1",
		UserID:             "usr_1",
		MemberID:           "mem_1",
		ScopeID:            "tsk_1",
		ScopeType:          unreadservice.ScopeTypeTask,
		Count:              0,
		ManualUnread:       false,
		HasUnread:          false,
		WorkspaceHasUnread: false,
		ServingState:       unreadservice.ServingStateReady,
	}}, sink.unreadUpdated)
}

func TestConversationMarkUnreadUsesRuntimeManualState(t *testing.T) {
	gin.SetMode(gin.TestMode)
	unread := &fakeUnreadRuntime{
		states: map[string]unreadservice.ScopeUnreadState{
			"chn_1": {ScopeID: "chn_1", ManualUnread: true, ServingState: unreadservice.ServingStateReady},
		},
		workspaceHasUnread: true,
	}
	sink := &recordingUnreadSink{}
	handler := &ConversationsHandler{
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:         message.ConversationKindChannel,
			ScopeID:      "chn_1",
			UnreadCount:  99,
			ManualUnread: false,
		}},
		unreadRuntime: unread,
		unreadSink:    sink,
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionMarkUnread}, gin.Param{Key: "conversationId", Value: "channel:chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	item := response.Body.(dto.ConversationItem)
	require.Equal(t, int64(0), item.UnreadCount)
	require.True(t, item.ManualUnread)
	require.Equal(t, []unreadservice.ManualUnreadChanged{{
		WorkspaceID:  "wsp_1",
		MemberID:     "mem_1",
		ScopeID:      "chn_1",
		ManualUnread: true,
	}}, unread.manualEvents)
	require.Equal(t, []unreadservice.UnreadDelta{{
		WorkspaceID:                      "wsp_1",
		UserID:                           "usr_1",
		MemberID:                         "mem_1",
		ScopeID:                          "chn_1",
		ScopeType:                        unreadservice.ScopeTypeChannel,
		Count:                            0,
		ManualUnread:                     true,
		HasUnread:                        true,
		WorkspaceHasUnread:               true,
		WorkspaceUnreadConversationCount: 1,
		ServingState:                     unreadservice.ServingStateReady,
	}}, sink.unreadUpdated)
}

func TestConversationSuppressedScopeKeepsWorkspaceAggregate(t *testing.T) {
	gin.SetMode(gin.TestMode)
	unread := &fakeUnreadRuntime{workspaceHasUnread: true}
	sink := &recordingUnreadSink{}
	handler := &ConversationsHandler{
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:         message.ConversationKindChannel,
			ScopeID:      "chn_1",
			UnreadCount:  3,
			ManualUnread: true,
		}},
		unreadRuntime: unread,
		unreadSink:    sink,
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionMute}, gin.Param{Key: "conversationId", Value: "channel:chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	item := response.Body.(dto.ConversationItem)
	require.Equal(t, int64(0), item.UnreadCount)
	require.False(t, item.ManualUnread)
	require.Equal(t, []unreadservice.ScopeInvalidated{{
		WorkspaceID: "wsp_1",
		ScopeID:     "chn_1",
		MemberIDs:   []string{"mem_1"},
		Reason:      "conversation_action",
	}}, unread.invalidations)
	require.Equal(t, []unreadservice.UnreadDelta{{
		WorkspaceID:        "wsp_1",
		UserID:             "usr_1",
		MemberID:           "mem_1",
		ScopeID:            "chn_1",
		ScopeType:          unreadservice.ScopeTypeChannel,
		Count:              0,
		ManualUnread:       false,
		HasUnread:          false,
		WorkspaceHasUnread: true,
		ServingState:       unreadservice.ServingStateStale,
	}}, sink.unreadUpdated)
}

func TestConversationMarkReadIgnoresStreamAppendFailure(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &ConversationsHandler{
		streamWriter: &recordingStreamWriter{memberErr: errors.New("append failed")},
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:        message.ConversationKindChannel,
			ScopeID:     "chn_1",
			Title:       "general",
			UnreadCount: 0,
		}},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionMarkRead}, gin.Param{Key: "conversationId", Value: "channel:chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
}

func TestConversationHideFallsBackToMemberLocalIndexAfterSourceDeletion(t *testing.T) {
	gin.SetMode(gin.TestMode)
	archivedAt := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC)
	streamWriter := &recordingStreamWriter{archived: storeim.ConversationIndexItem{
		WorkspaceID:    "wsp_1",
		MemberID:       "mem_1",
		ConversationID: "serviceAccount:sasc_deleted_service",
		ScopeID:        "sasc_deleted_service",
		Kind:           message.ConversationKindServiceAccount,
		SortAt:         archivedAt,
		ArchivedAt:     &archivedAt,
		Projection: map[string]any{
			"id":            "serviceAccount:sasc_deleted_service",
			"scopeId":       "sasc_deleted_service",
			"kind":          "serviceAccount",
			"route":         "/inbox/service-accounts/sacct_deleted_service",
			"title":         "Deleted Service",
			"peerType":      "service",
			"peerId":        "sacct_deleted_service",
			"sidebarHidden": true,
			"archived":      true,
		},
	}}
	handler := &ConversationsHandler{
		streamWriter:  streamWriter,
		conversations: fakeConversationStore{actionErr: message.ErrNotFound},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionHide}, gin.Param{Key: "conversationId", Value: "serviceAccount:sasc_deleted_service"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	item, ok := response.Body.(dto.ConversationItem)
	require.True(t, ok)
	require.Equal(t, "serviceAccount:sasc_deleted_service", item.ID)
	require.True(t, item.SidebarHidden)
	require.True(t, item.Archived)
	require.Len(t, streamWriter.archiveRequests, 1)
	require.Equal(t, "serviceAccount:sasc_deleted_service", streamWriter.archiveRequests[0].ConversationID)
	require.Empty(t, streamWriter.memberRequests)
}

func TestConversationHideBuildsThreadRouteFromArchivedScope(t *testing.T) {
	gin.SetMode(gin.TestMode)
	archivedAt := time.Date(2026, 7, 17, 12, 0, 0, 0, time.UTC)
	streamWriter := &recordingStreamWriter{archived: storeim.ConversationIndexItem{
		WorkspaceID:    "wsp_1",
		MemberID:       "mem_1",
		ConversationID: "thread:thr_archived",
		ScopeID:        "thr_archived",
		Kind:           message.ConversationKindThread,
		SortAt:         archivedAt,
		ArchivedAt:     &archivedAt,
		Projection: map[string]any{
			"id":            "thread:thr_archived",
			"scopeId":       "thr_archived",
			"kind":          "replyThread",
			"route":         "/inbox/invalid",
			"title":         "Thread",
			"sidebarHidden": true,
			"archived":      true,
		},
	}}
	handler := &ConversationsHandler{
		streamWriter:  streamWriter,
		conversations: fakeConversationStore{actionErr: message.ErrNotFound},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionHide}, gin.Param{Key: "conversationId", Value: "thread:thr_archived"}))

	require.NoError(t, err)
	item := response.Body.(dto.ConversationItem)
	require.Equal(t, "/inbox/thread/thr_archived", item.Route)
}

func TestConversationHideMissingSourceAndMemberIndexReturnsNotFound(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &ConversationsHandler{
		streamWriter:  &recordingStreamWriter{archiveErr: sql.ErrNoRows},
		conversations: fakeConversationStore{actionErr: message.ErrNotFound},
	}

	_, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionHide}, gin.Param{Key: "conversationId", Value: "serviceAccount:sasc_missing"}))
	require.ErrorIs(t, err, message.ErrNotFound)
}

func TestConversationHideIncompleteLegacyProjectionStillReturnsArchived(t *testing.T) {
	gin.SetMode(gin.TestMode)
	archivedAt := time.Date(2026, 7, 14, 12, 15, 0, 0, time.UTC)
	handler := &ConversationsHandler{
		streamWriter: &recordingStreamWriter{archived: storeim.ConversationIndexItem{
			WorkspaceID:    "wsp_1",
			MemberID:       "mem_1",
			ConversationID: "serviceAccount:sasc_legacy",
			ScopeID:        "sasc_legacy",
			Kind:           message.ConversationKindServiceAccount,
			SortAt:         archivedAt,
			ArchivedAt:     &archivedAt,
			Projection:     map[string]any{"peerType": "service", "peerId": "sacct_legacy"},
		}},
		conversations: fakeConversationStore{actionErr: message.ErrNotFound},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionHide}, gin.Param{Key: "conversationId", Value: "serviceAccount:sasc_legacy"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	item := response.Body.(dto.ConversationItem)
	require.Equal(t, "serviceAccount:sasc_legacy", item.ID)
	require.Equal(t, "serviceAccount", item.Kind)
	require.Equal(t, "sasc_legacy", item.Title)
	require.Equal(t, "/inbox/service-accounts/sacct_legacy", item.Route)
	require.True(t, item.Archived)
}

func TestConversationHideFallsBackWhenSourceDisappearsBeforeStreamProjection(t *testing.T) {
	gin.SetMode(gin.TestMode)
	archivedAt := time.Date(2026, 7, 14, 12, 30, 0, 0, time.UTC)
	streamWriter := &recordingStreamWriter{
		memberErr: sql.ErrNoRows,
		archived: storeim.ConversationIndexItem{
			WorkspaceID:    "wsp_1",
			MemberID:       "mem_1",
			ConversationID: "serviceAccount:sasc_deleted_during_action",
			ScopeID:        "sasc_deleted_during_action",
			Kind:           message.ConversationKindServiceAccount,
			SortAt:         archivedAt,
			ArchivedAt:     &archivedAt,
			Projection: map[string]any{
				"id":            "serviceAccount:sasc_deleted_during_action",
				"scopeId":       "sasc_deleted_during_action",
				"kind":          "serviceAccount",
				"route":         "/inbox/service-accounts/sacct_deleted_during_action",
				"title":         "Deleted During Action",
				"peerType":      "service",
				"peerId":        "sacct_deleted_during_action",
				"sidebarHidden": true,
				"archived":      true,
			},
		},
	}
	handler := &ConversationsHandler{
		streamWriter: streamWriter,
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:          message.ConversationKindServiceAccount,
			ScopeID:       "sasc_deleted_during_action",
			Title:         "Deleted During Action",
			PeerType:      "service",
			PeerID:        "sacct_deleted_during_action",
			SidebarHidden: true,
			Archived:      true,
		}},
	}

	response, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionHide}, gin.Param{Key: "conversationId", Value: "serviceAccount:sasc_deleted_during_action"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.memberRequests, 1, "the source-backed projection was attempted first")
	require.Len(t, streamWriter.archiveRequests, 1, "the exact index action must finish the delete race")
}

func TestConversationNonReadActionPreservesStreamAppendFailure(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &ConversationsHandler{
		streamWriter: &recordingStreamWriter{memberErr: errors.New("append failed")},
		conversations: fakeConversationStore{item: message.ConversationItem{
			Kind:    message.ConversationKindChannel,
			ScopeID: "chn_1",
			Title:   "general",
		}},
	}

	_, err := handler.ActionJSON(imWorkspaceJSON(conversationActionRequest{Action: message.ConversationActionPin}, gin.Param{Key: "conversationId", Value: "channel:chn_1"}))
	require.ErrorContains(t, err, "append failed")
}

func TestConversationBulkActionAppendsWorkspaceStreamEvent(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	handler := &ConversationsHandler{
		streamWriter: streamWriter,
		conversations: fakeConversationStore{bulkResult: message.ConversationBulkActionResult{
			UpdatedCount: 3,
		}},
	}

	response, err := handler.BulkAction(imWorkspaceJSON(conversationsBulkActionRequest{
		Action: message.ConversationActionMarkRead,
		Kinds:  []string{"channel", "replyThread"},
	}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.workspaceRequests, 1)
	recorded := streamWriter.workspaceRequests[0]
	require.Equal(t, "wsp_1", recorded.workspaceID)
	require.Equal(t, "mem_1", recorded.memberID)
	require.Equal(t, imEventTypeConversationBulkRead, recorded.event.Type)
	require.Equal(t, message.ConversationActionMarkRead, recorded.event.Payload["action"])
	require.Contains(t, recorded.event.Payload, "result")
}

func TestConversationBulkReadFeedsUnreadRuntime(t *testing.T) {
	gin.SetMode(gin.TestMode)
	unread := &fakeUnreadRuntime{states: map[string]unreadservice.ScopeUnreadState{
		"chn_1": {ScopeID: "chn_1", ServingState: unreadservice.ServingStateReady},
		"thr_1": {ScopeID: "thr_1", ServingState: unreadservice.ServingStateReady},
	}}
	sink := &recordingUnreadSink{}
	handler := &ConversationsHandler{
		conversations: fakeConversationStore{bulkResult: message.ConversationBulkActionResult{
			UpdatedCount: 2,
			ReadScopes: []message.ConversationReadScope{
				{ScopeID: "chn_1", Kind: message.ConversationKindChannel, LastReadSeq: 5},
				{ScopeID: "thr_1", Kind: message.ConversationKindThread, LastReadSeq: 7},
			},
		}},
		unreadRuntime: unread,
		unreadSink:    sink,
	}

	response, err := handler.BulkAction(imWorkspaceJSON(conversationsBulkActionRequest{Action: message.ConversationActionMarkRead}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	result := response.Body.(dto.ConversationsBulkActionResponse)
	require.Equal(t, int64(2), result.UpdatedCount)
	require.Equal(t, []unreadservice.ReadCursorAdvanced{
		{WorkspaceID: "wsp_1", MemberID: "mem_1", ScopeID: "chn_1", LastReadSeq: 5},
		{WorkspaceID: "wsp_1", MemberID: "mem_1", ScopeID: "thr_1", LastReadSeq: 7},
	}, unread.readEvents)
	require.Equal(t, []unreadservice.ActivityReadCursorAdvanced{
		{WorkspaceID: "wsp_1", MemberID: "mem_1", ScopeID: "chn_1"},
		{WorkspaceID: "wsp_1", MemberID: "mem_1", ScopeID: "thr_1"},
	}, unread.activityReadEvents)
	require.Len(t, sink.unreadUpdated, 2)
	require.Equal(t, "chn_1", sink.unreadUpdated[0].ScopeID)
	require.Equal(t, "thr_1", sink.unreadUpdated[1].ScopeID)
}

func imWorkspaceRequest(params ...gin.Param) api.WorkspaceScopedRequest {
	ctx, _ := gin.CreateTestContext(httptest.NewRecorder())
	ctx.Request = httptest.NewRequest(http.MethodPost, "/", nil)
	ctx.Params = params
	return api.WorkspaceScopedRequest{
		AuthenticatedRequest: api.AuthenticatedRequest{Gin: ctx, User: identity.AuthenticatedUser{User: identity.User{ID: "usr_1"}}},
		Workspace: workspace.Summary{
			WorkspaceID:  "wsp_1",
			MemberID:     "mem_1",
			MemberUserID: "usr_1",
		},
	}
}

func imWorkspaceJSON[T any](body T, params ...gin.Param) api.WorkspaceScopedJSON[T] {
	return api.WorkspaceScopedJSON[T]{WorkspaceScopedRequest: imWorkspaceRequest(params...), Body: body}
}

type fakeReactionStore struct {
	changed     bool
	channelID   string
	channelType string
	reactions   []message.MessageReactionAggregate
}

func (store fakeReactionStore) GetMessageReactionsForUser(context.Context, string, string, string) ([]message.MessageReactionAggregate, error) {
	return store.reactions, nil
}
func (store fakeReactionStore) AddReactionForUser(context.Context, string, string, string, string) (message.AddReactionResult, error) {
	return message.AddReactionResult{
		Changed:     store.changed,
		ChannelID:   store.channelID,
		ChannelType: store.channelType,
	}, nil
}
func (store fakeReactionStore) RemoveReactionForUser(context.Context, string, string, string, string) (message.RemoveReactionResult, error) {
	return message.RemoveReactionResult{
		Changed:     store.changed,
		ChannelID:   store.channelID,
		ChannelType: store.channelType,
	}, nil
}

type fakeRecallWriter struct {
	result message.AppMessageRecallResult
	err    error
}

func (writer fakeRecallWriter) ValidateHumanMessage(context.Context, identity.User, string, string, string, messageservice.SendOptions) error {
	return nil
}

func (writer fakeRecallWriter) SendHumanMessage(context.Context, identity.User, string, string, string, []string, messageservice.SendOptions) (message.AppMessage, message.MessageDelivery, error) {
	return message.AppMessage{}, message.MessageDelivery{}, errors.New("unexpected send")
}

func (writer fakeRecallWriter) RecallHumanMessage(context.Context, identity.User, string, string, string) (message.AppMessageRecallResult, error) {
	return writer.result, writer.err
}

type rewindNoticeSendCall struct {
	user          identity.User
	workspaceID   string
	scopeID       string
	content       string
	attachmentIDs []string
	options       messageservice.SendOptions
}

type fakeRewindNoticeWriter struct {
	calls            []rewindNoticeSendCall
	err              error
	idempotentReplay bool
}

func (writer *fakeRewindNoticeWriter) ValidateHumanMessage(context.Context, identity.User, string, string, string, messageservice.SendOptions) error {
	return nil
}

func (writer *fakeRewindNoticeWriter) SendHumanMessage(_ context.Context, user identity.User, workspaceID string, scopeID string, content string, attachmentIDs []string, options messageservice.SendOptions) (message.AppMessage, message.MessageDelivery, error) {
	writer.calls = append(writer.calls, rewindNoticeSendCall{
		user:          user,
		workspaceID:   workspaceID,
		scopeID:       scopeID,
		content:       content,
		attachmentIDs: attachmentIDs,
		options:       options,
	})
	if writer.err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, writer.err
	}
	createdAt := time.Date(2026, 6, 29, 3, 58, 0, 0, time.UTC)
	displayName := firstNonEmptyString(user.DisplayName, user.Name, user.Email, user.ID)
	profile := message.AuthorProfile{
		ActorType:   message.AuthorActorTypeHuman,
		PrincipalID: user.ID,
		MemberID:    "mem_1",
		DisplayName: displayName,
	}
	appMessage := message.AppMessage{
		ID:              "msg_rewind_notice",
		Seq:             10,
		ChannelID:       scopeID,
		ClientMessageID: options.ClientMessageID,
		AuthorProfile:   profile,
		Content:         content,
		CreatedAt:       createdAt,
		MessageType:     message.MessageTypeSystem,
		Metadata:        message.CopyMessageMetadata(options.Metadata),
	}
	delivery := message.MessageDelivery{
		ThreadID:         scopeID,
		Seq:              appMessage.Seq,
		GlobalSeq:        appMessage.Seq,
		MessageID:        appMessage.ID,
		ClientMessageID:  options.ClientMessageID,
		AppMessage:       &appMessage,
		IdempotentReplay: writer.idempotentReplay,
		Message: message.ReceiveMessage{
			MessageID:     appMessage.ID,
			TargetKey:     scopeID,
			TargetSeq:     appMessage.Seq,
			AuthorProfile: profile,
			MessageType:   appMessage.MessageType,
			Content:       appMessage.Content,
			Metadata:      message.CopyMessageMetadata(appMessage.Metadata),
			Timestamp:     createdAt,
		},
	}
	return appMessage, delivery, nil
}

func TestMessageRecallAppendsRecalledEventWhenChanged(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	unread := &fakeUnreadRuntime{invalidationDeltas: []unreadservice.UnreadDelta{{
		WorkspaceID:        "wsp_1",
		UserID:             "usr_1",
		MemberID:           "mem_1",
		ScopeID:            "chn_1",
		ScopeType:          unreadservice.ScopeTypeChannel,
		Count:              0,
		ManualUnread:       false,
		HasUnread:          false,
		WorkspaceHasUnread: false,
		ServingState:       unreadservice.ServingStateStale,
	}}}
	sink := &recordingUnreadSink{}
	recalledAt := time.Date(2026, 6, 26, 9, 30, 0, 0, time.UTC)
	handler := &MessagesHandler{
		streamWriter:  streamWriter,
		unreadRuntime: unread,
		unreadSink:    sink,
		writer: fakeRecallWriter{result: message.AppMessageRecallResult{
			Message: message.AppMessage{
				ID:        "msg_1",
				Seq:       5,
				ChannelID: "chn_1",
				Content:   "Alice 撤回了一条消息",
				Metadata:  message.BuildMessageRecallMetadata("mem_1", recalledAt.Format(time.RFC3339Nano), "Alice 撤回了一条消息"),
				CreatedAt: recalledAt,
				AuthorProfile: message.AuthorProfile{
					ActorType:   message.AuthorActorTypeHuman,
					PrincipalID: "usr_1",
					MemberID:    "mem_1",
					DisplayName: "Alice",
				},
			},
			Changed:                true,
			IsLatestVisibleMessage: true,
		}},
	}

	response, err := handler.Recall(imWorkspaceJSON(messageRecallRequest{},
		gin.Param{Key: "scopeId", Value: "chn_1"},
		gin.Param{Key: "messageId", Value: "msg_1"},
	))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	body := response.Body.(dto.AppMessageRecallResponse)
	require.True(t, body.OK)
	require.Equal(t, "recall", body.Mode)
	require.True(t, body.Changed)
	require.NotNil(t, body.Message)
	require.Equal(t, "Alice 撤回了一条消息", body.Message.Content)

	require.Len(t, streamWriter.scopeRequests, 2)
	recallRequest := streamWriter.scopeRequests[0]
	require.Equal(t, "wsp_1", recallRequest.WorkspaceID)
	require.Equal(t, "chn_1", recallRequest.ScopeID)
	require.Equal(t, imEventTypeMessageRecalled, recallRequest.Event.Type)
	require.Equal(t, "msg_1", recallRequest.Event.EntityID)
	require.Equal(t, "msg_1", recallRequest.Event.Payload["messageId"])
	require.Contains(t, recallRequest.Event.Payload, "tombstone")
	require.NotNil(t, recallRequest.Conversation)
	require.Equal(t, "chn_1", recallRequest.Conversation.ScopeID)

	pinRequest := streamWriter.scopeRequests[1]
	require.Equal(t, imEventTypeMessagePinsInvalidated, pinRequest.Event.Type)
	require.Equal(t, []string{"msg_1"}, pinRequest.Event.Payload["messageIds"])
	require.Equal(t, []unreadservice.ScopeInvalidated{{
		WorkspaceID: "wsp_1",
		ScopeID:     "chn_1",
		Reason:      "message_recall",
	}}, unread.invalidations)
	require.Equal(t, unread.invalidationDeltas, sink.unreadUpdated)
}

func TestMessageRecallOfThreadMessageInvalidatesThreadPinScope(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	recalledAt := time.Date(2026, 6, 26, 9, 45, 0, 0, time.UTC)
	handler := &MessagesHandler{
		streamWriter: streamWriter,
		writer: fakeRecallWriter{result: message.AppMessageRecallResult{
			Message: message.AppMessage{
				ID:              "msg_thread_1",
				Seq:             2,
				ChannelID:       "thr_1",
				ChannelType:     "thread",
				ParentChannelID: "chn_1",
				Content:         "Alice 撤回了一条消息",
				Metadata:        message.BuildMessageRecallMetadata("mem_1", recalledAt.Format(time.RFC3339Nano), "Alice 撤回了一条消息"),
				CreatedAt:       recalledAt,
				AuthorProfile: message.AuthorProfile{
					ActorType:   message.AuthorActorTypeHuman,
					PrincipalID: "usr_1",
					MemberID:    "mem_1",
					DisplayName: "Alice",
				},
			},
			Changed: true,
		}},
	}

	_, err := handler.Recall(imWorkspaceJSON(messageRecallRequest{},
		gin.Param{Key: "scopeId", Value: "thr_1"},
		gin.Param{Key: "messageId", Value: "msg_thread_1"},
	))
	require.NoError(t, err)
	require.Len(t, streamWriter.scopeRequests, 2)
	require.Equal(t, imEventTypeMessagePinsInvalidated, streamWriter.scopeRequests[1].Event.Type)
	require.Equal(t, "thr_1", streamWriter.scopeRequests[1].ScopeID)
}

type fakeRecallRewindStore struct {
	plan      rewind.Plan
	operation string
	prepared  bool
	validated bool
	applied   bool
}

func (store *fakeRecallRewindStore) PrepareRewindForUser(context.Context, string, string, string) (rewind.Plan, error) {
	store.prepared = true
	return store.plan, nil
}

func (store *fakeRecallRewindStore) BeginRewindOperation(context.Context, rewind.Plan) (string, error) {
	if store.operation == "" {
		store.operation = "rewind_op_1"
	}
	return store.operation, nil
}

func (store *fakeRecallRewindStore) MarkRewindOperationFailed(context.Context, string, string) error {
	return nil
}

func (store *fakeRecallRewindStore) MarkRewindOperationRepairRequired(context.Context, string, string) error {
	return nil
}

func (store *fakeRecallRewindStore) ValidateRewindPlan(context.Context, rewind.Plan) error {
	store.validated = true
	return nil
}

func (store *fakeRecallRewindStore) ApplyRewind(context.Context, string, rewind.Plan) error {
	store.applied = true
	return nil
}

type fakeRewindHub struct {
	interruptCalls []string
	restoreCalls   []runtimecontrol.RestoreAgentRecoveryPointRequest
	restoreMachine []string
}

func (hub *fakeRewindHub) NotifyAgentMessage(context.Context, message.AgentRecipient, int64, message.ReceiveMessage) error {
	return nil
}

func (hub *fakeRewindHub) RestoreAgentRecoveryPoint(_ context.Context, machineID string, request runtimecontrol.RestoreAgentRecoveryPointRequest) error {
	hub.restoreMachine = append(hub.restoreMachine, machineID)
	hub.restoreCalls = append(hub.restoreCalls, request)
	return nil
}

func (hub *fakeRewindHub) InterruptAgentRun(machineID string, agentID string) error {
	hub.interruptCalls = append(hub.interruptCalls, machineID+"/"+agentID)
	return nil
}

func TestRewindMessageInterruptsPendingRuntimeAgentsBeforeApply(t *testing.T) {
	streamWriter := &recordingStreamWriter{}
	noticeWriter := &fakeRewindNoticeWriter{}
	store := &fakeRecallRewindStore{plan: rewind.Plan{
		WorkspaceID:      "wsp_1",
		ScopeID:          "chn_1",
		BroadcastScopeID: "chn_1",
		TargetMessageID:  "msg_2",
		TargetSeq:        2,
		SuffixMessageIDs: []string{"msg_2"},
		RewoundScopes:    []rewind.RewoundScope{{ScopeID: "chn_1", LatestSeq: 1}},
		RelatedAgents:    nil,
		PendingInterruptAgents: []rewind.AgentPlan{{
			AgentID:             "agt_codex",
			MachineID:           "mac_codex",
			Runtime:             "codex",
			FirstTouchMessageID: "msg_2",
			FirstTouchDelivery:  true,
		}},
	}}
	hub := &fakeRewindHub{}
	handler := &RewindHandler{
		store:        store,
		hub:          hub,
		streamWriter: streamWriter,
		noticeWriter: noticeWriter,
	}

	_, err := handler.rewindMessage(context.Background(), identity.User{ID: "usr_1", DisplayName: "User"}, "wsp_1", "msg_2", "")

	require.NoError(t, err)
	require.Equal(t, []string{"mac_codex/agt_codex"}, hub.interruptCalls)
	require.Empty(t, hub.restoreCalls)
	require.True(t, store.validated)
	require.True(t, store.applied)
}

func TestMessageRecallWithRewindConversationUsesRewindPath(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	noticeWriter := &fakeRewindNoticeWriter{}
	store := &fakeRecallRewindStore{plan: rewind.Plan{
		WorkspaceID:         "wsp_1",
		ScopeID:             "chn_1",
		BroadcastScopeID:    "chn_1",
		TargetMessageID:     "msg_1",
		TargetSeq:           9,
		SuffixMessageIDs:    []string{"msg_1", "msg_2"},
		RewoundScopes:       []rewind.RewoundScope{{ScopeID: "chn_1", LatestSeq: 8}},
		RelatedAgents:       nil,
		CascadeReplyThreads: []rewind.CascadeReplyThread{{ScopeID: "thr_1", ParentMessageID: "msg_2", MessageIDs: []string{"msg_r1"}}},
	}}
	handler := &MessagesHandler{
		streamWriter: streamWriter,
		writer:       fakeRecallWriter{},
		rewind: &RewindHandler{
			store:        store,
			streamWriter: streamWriter,
			noticeWriter: noticeWriter,
		},
	}

	request := imWorkspaceJSON(messageRecallRequest{RewindConversation: true},
		gin.Param{Key: "scopeId", Value: "chn_1"},
		gin.Param{Key: "messageId", Value: "msg_1"},
	)
	request.User.User.DisplayName = "Test User"
	response, err := handler.Recall(request)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.True(t, store.prepared)
	require.True(t, store.validated)
	require.True(t, store.applied)

	body := response.Body.(dto.AppMessageRecallResponse)
	require.True(t, body.OK)
	require.Equal(t, "rewind", body.Mode)
	require.NotNil(t, body.Rewind)
	require.Equal(t, "chn_1", body.Rewind.ScopeID)
	require.Equal(t, "msg_1", body.Rewind.MessageID)
	require.Equal(t, int64(9), body.Rewind.FromSeq)
	require.Equal(t, []string{"msg_1", "msg_2"}, body.Rewind.RewoundMessageIDs)

	require.Len(t, noticeWriter.calls, 1)
	noticeCall := noticeWriter.calls[0]
	require.Equal(t, "usr_1", noticeCall.user.ID)
	require.Equal(t, "wsp_1", noticeCall.workspaceID)
	require.Equal(t, "chn_1", noticeCall.scopeID)
	require.Equal(t, "system.rewind.conversation", noticeCall.content)
	require.Empty(t, noticeCall.attachmentIDs)
	require.Equal(t, message.MessageTypeSystem, noticeCall.options.MessageType)
	require.Equal(t, map[string]any{
		message.SystemNoticeMetadataKey: map[string]any{
			"kind":             message.SystemNoticeKindRewindConversation,
			"actorDisplayName": "Test User",
		},
	}, noticeCall.options.Metadata)
	require.Equal(t, "rewind-notice:rewind_op_1", noticeCall.options.ClientMessageID)

	require.Len(t, streamWriter.scopeRequests, 4)
	require.Equal(t, "conversation.rewound", streamWriter.scopeRequests[0].Event.Type)
	require.Equal(t, imEventTypeMessagePinsInvalidated, streamWriter.scopeRequests[1].Event.Type)
	require.Equal(t, "chn_1", streamWriter.scopeRequests[1].ScopeID)
	require.Equal(t, imEventTypeMessagePinsInvalidated, streamWriter.scopeRequests[2].Event.Type)
	require.Equal(t, "thr_1", streamWriter.scopeRequests[2].ScopeID)
	require.Equal(t, imEventTypeMessageCreated, streamWriter.scopeRequests[3].Event.Type)
	require.Equal(t, "chn_1", streamWriter.scopeRequests[3].ScopeID)
	require.Equal(t, "msg_rewind_notice", streamWriter.scopeRequests[3].Event.EntityID)
	require.Equal(t, true, streamWriter.scopeRequests[3].Event.Payload["suppressBrowserNotification"])
}

func TestReactionAddAppendsScopeStreamEventWhenChanged(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	handler := &ReactionsHandler{
		streamWriter: streamWriter,
		reactions: fakeReactionStore{
			changed:     true,
			channelID:   "chn_1",
			channelType: message.ConversationKindChannel,
			reactions:   []message.MessageReactionAggregate{{Emoji: "👍", Count: 2}},
		},
	}

	response, err := handler.Add(imWorkspaceRequest(
		gin.Param{Key: "scopeId", Value: "chn_1"},
		gin.Param{Key: "messageId", Value: "msg_1"},
		gin.Param{Key: "emoji", Value: "👍"},
	))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.scopeRequests, 1)
	request := streamWriter.scopeRequests[0]
	require.Equal(t, "wsp_1", request.WorkspaceID)
	require.Equal(t, "chn_1", request.ScopeID)
	require.Equal(t, imEventTypeReactionChanged, request.Event.Type)
	require.Equal(t, "msg_1", request.Event.EntityID)
	require.Equal(t, "msg_1", request.Event.Payload["messageId"])
	require.Contains(t, request.Event.Payload, "reactions")
}

type fakeDMStore struct {
	channel message.DMChannel
}

func (store fakeDMStore) ListDMChannelsForUserWithLimit(context.Context, string, string, int) ([]message.DMChannel, error) {
	return []message.DMChannel{store.channel}, nil
}
func (store fakeDMStore) GetDMChannelForUser(context.Context, string, string, string) (message.DMChannel, error) {
	return store.channel, nil
}
func (store fakeDMStore) OpenDMChannelWithAgent(context.Context, string, string, string) (message.DMChannel, error) {
	return store.channel, nil
}
func (store fakeDMStore) OpenDMChannelWithUser(context.Context, string, string, string) (message.DMChannel, error) {
	return store.channel, nil
}

func TestDMOpenAppendsMemberStreamEventAndProjection(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	lastMessageAt := time.Date(2026, 6, 10, 11, 0, 0, 0, time.UTC)
	handler := &DMHandler{
		streamWriter: streamWriter,
		dms: fakeDMStore{channel: message.DMChannel{
			ID:            "dm_1",
			PeerType:      "user",
			PeerID:        "usr_peer",
			PeerName:      "Peer",
			LastMessageAt: &lastMessageAt,
		}},
	}

	response, err := handler.Open(imWorkspaceJSON(dto.OpenDMRequest{UserID: "usr_peer"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, "wsp_1", recorded.workspaceID)
	require.Equal(t, "mem_1", recorded.memberID)
	require.Equal(t, "dm_1", recorded.scopeID)
	require.Equal(t, imEventTypeDMOpened, recorded.event.Type)
	require.NotNil(t, recorded.conversation)
	require.Equal(t, "dm:dm_1", recorded.conversation.ConversationID)
	require.Equal(t, message.ConversationKindDM, recorded.conversation.Kind)
	require.Equal(t, "dm_1", recorded.event.Payload["channelId"])
	require.Contains(t, recorded.event.Payload, "channel")
}

type fakeOwnerContextStore struct{}

func (fakeOwnerContextStore) DismissPendingOwnerContextForUser(context.Context, string, string, string, int64) error {
	return nil
}

type fakeScopeNotificationStore struct {
	panel message.ScopeNotificationPanel
}

func (store fakeScopeNotificationStore) GetChannelNotificationPanelForUser(context.Context, string, string, string, string, int) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}
func (store fakeScopeNotificationStore) SetChannelNotificationSelfStateForUser(context.Context, string, string, string, string) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}
func (store fakeScopeNotificationStore) SetChannelNotificationAgentWakeStateForUser(context.Context, string, string, string, string, string) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}
func (store fakeScopeNotificationStore) GetThreadNotificationPanelForUser(context.Context, string, string, string, string, int) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}
func (store fakeScopeNotificationStore) SetThreadNotificationSelfStateForUser(context.Context, string, string, string, string) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}
func (store fakeScopeNotificationStore) SetThreadNotificationAgentWakeStateForUser(context.Context, string, string, string, string, string) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}
func (store fakeScopeNotificationStore) SetNotificationAgentsWakeStateForUser(context.Context, string, string, string, string, string) (message.ScopeNotificationPanel, error) {
	return store.panel, nil
}

func TestOwnerContextDismissAppendsMemberStreamEvent(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	handler := &ScopeExtrasHandler{
		streamWriter:  streamWriter,
		ownerContexts: fakeOwnerContextStore{},
	}

	response, err := handler.DismissOwnerContext(imWorkspaceJSON(dto.ChannelDismissOwnerContextRequest{DismissedThroughGlobalSeq: 42}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, "chn_1", recorded.scopeID)
	require.Equal(t, imEventTypeOwnerContextDismissed, recorded.event.Type)
	require.Equal(t, int64(42), recorded.event.Payload["dismissedThroughSeq"])
}

func TestScopeNotificationsUseConversationKindForThreadChannels(t *testing.T) {
	gin.SetMode(gin.TestMode)
	handler := &ScopeExtrasHandler{
		channelNotifications: fakeScopeNotificationStore{
			panel: message.ScopeNotificationPanel{ScopeID: "chn_thread", ScopeType: "channel"},
		},
		threadNotifications: fakeScopeNotificationStore{
			panel: message.ScopeNotificationPanel{ScopeID: "chn_thread", ScopeType: "thread"},
		},
		conversations: fakeConversationStore{item: message.ConversationItem{
			ScopeID: "chn_thread",
			Kind:    message.ConversationKindThread,
		}},
	}

	response, err := handler.NotificationsGet(imWorkspaceRequest(gin.Param{Key: "scopeId", Value: "chn_thread"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	panel := response.Body.(dto.ScopeNotificationPanel)
	require.Equal(t, "chn_thread", panel.ScopeID)
	require.Equal(t, "thread", panel.ScopeType)
}

func TestNotificationSelfSetAppendsMemberStreamEventAndClearsUnreadBadge(t *testing.T) {
	gin.SetMode(gin.TestMode)
	streamWriter := &recordingStreamWriter{}
	unread := &fakeUnreadRuntime{workspaceHasUnread: true}
	sink := &recordingUnreadSink{}
	panel := message.ScopeNotificationPanel{ScopeID: "chn_1", ScopeType: "channel"}
	handler := &ScopeExtrasHandler{
		streamWriter:         streamWriter,
		channelNotifications: fakeScopeNotificationStore{panel: panel},
		conversations: fakeConversationStore{item: message.ConversationItem{
			ScopeID: "chn_1",
			Kind:    message.ConversationKindChannel,
		}},
		unreadRuntime: unread,
		unreadSink:    sink,
	}

	response, err := handler.NotificationsSelfSet(imWorkspaceJSON(dto.NotificationStateRequest{State: "mute"}, gin.Param{Key: "scopeId", Value: "chn_1"}))
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.Status)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, "chn_1", recorded.scopeID)
	require.Equal(t, imEventTypeNotificationUpdated, recorded.event.Type)
	require.Equal(t, "self", recorded.event.Payload["subject"])
	require.Contains(t, recorded.event.Payload, "panel")
	require.Equal(t, []unreadservice.ScopeInvalidated{{
		WorkspaceID: "wsp_1",
		ScopeID:     "chn_1",
		MemberIDs:   []string{"mem_1"},
		Reason:      "channel_notification_self",
	}}, unread.invalidations)
	require.Equal(t, []unreadservice.UnreadDelta{{
		WorkspaceID:        "wsp_1",
		UserID:             "usr_1",
		MemberID:           "mem_1",
		ScopeID:            "chn_1",
		ScopeType:          unreadservice.ScopeTypeChannel,
		Count:              0,
		ManualUnread:       false,
		HasUnread:          false,
		WorkspaceHasUnread: true,
		ServingState:       unreadservice.ServingStateStale,
	}}, sink.unreadUpdated)
}