package fanout

import (
	"errors"
	"fmt"
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/store/message"
	unreadservice "github.com/tidbcloud/AgentTeam/internal/unread"
)

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

	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster("wsp_1", "tsk_1", unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "active"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "active"},
	}))
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(nil, realtime, nil).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)

	fanout.EmitCommittedHumanMessage(t.Context(), "wsp_1", message.AppMessage{
		ID:          "msg_task_unread",
		Seq:         8,
		ChannelID:   "tsk_1",
		ChannelType: "task",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			MemberID:    "mem_author",
			DisplayName: "Alice",
		},
		MessageType: message.MessageTypeTaskDiscussion,
	}, message.MessageDelivery{
		MessageID:               "msg_task_unread",
		Seq:                     8,
		HumanRecipientMemberIDs: []string{"mem_reader"},
		HumanRecipientsTargeted: true,
		Message: message.ReceiveMessage{
			MessageID:   "msg_task_unread",
			TargetKey:   "tsk_1",
			ChannelType: "task",
		},
	})

	require.Equal(t, []unreadservice.MessageStored{{
		WorkspaceID:             "wsp_1",
		ScopeID:                 "tsk_1",
		ScopeType:               unreadservice.ScopeTypeTask,
		Seq:                     8,
		AuthorMemberID:          "mem_author",
		NotifyPrincipalMemberID: "mem_author",
		MessageGeneration:       8,
		RecipientMemberIDs:      []string{"mem_reader"},
	}}, unread.messages)
	require.Len(t, realtime.unreadUpdated, 1)
	stateVersion := realtime.unreadUpdated[0].StateVersion
	realtime.unreadUpdated[0].StateVersion = 0
	require.Equal(t, []unreadservice.UnreadDelta{{
		WorkspaceID:                      "wsp_1",
		UserID:                           "usr_reader",
		MemberID:                         "mem_reader",
		ScopeID:                          "tsk_1",
		ScopeType:                        unreadservice.ScopeTypeTask,
		Count:                            1,
		HasUnread:                        true,
		WorkspaceHasUnread:               true,
		WorkspaceUnreadConversationCount: 1,
		ServingState:                     unreadservice.ServingStateReady,
	}}, realtime.unreadUpdated)
	require.Positive(t, stateVersion)
	require.True(t, runtime.WorkspaceHasUnread("wsp_1", "mem_reader"))
}

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

	const (
		workspaceID = "wsp_task_pending"
		scopeID     = "tsk_task_pending"
	)
	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	notifier := &capturingNotifier{}
	marker := &recordingScopeFanoutMarker{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).
		WithScopeFanoutMarker(marker).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)
	appMessage := taskLifecycleAppMessage(scopeID)
	delivery := taskLifecycleDelivery(scopeID)

	err = fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery)
	require.ErrorIs(t, err, errTargetedTaskUnreadPending)
	require.Len(t, realtime.messageNew, 1, "the durable browser projection precedes unread accounting")
	require.Empty(t, notifier.messages, "a cold unread roster must not wake the Agent")
	require.Empty(t, marker.ids, "pending unread must keep the message outbox incomplete")
	require.Empty(t, realtime.unreadUpdated)

	require.NoError(t, runtime.WarmScopeRoster(workspaceID, scopeID, unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "mute"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "mute"},
	}))

	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery))
	require.Len(t, realtime.messageNew, 1, "stable stream event ids make the retry idempotent")
	require.Len(t, notifier.messages, 1)
	require.Equal(t, []string{delivery.MessageID}, marker.ids)
	require.Len(t, realtime.unreadUpdated, 1)
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, "mem_reader", nil)[scopeID])
}

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

	const (
		workspaceID = "wsp_ordinary_pending"
		scopeID     = "chn_ordinary_pending"
	)
	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	notifier := &capturingNotifier{}
	marker := &recordingScopeFanoutMarker{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).
		WithScopeFanoutMarker(marker).
		WithUnreadRuntime(unread)
	appMessage := taskLifecycleAppMessage(scopeID)
	appMessage.ChannelType = message.ConversationKindChannel
	appMessage.MessageType = ""
	delivery := taskLifecycleDelivery(scopeID)
	delivery.HumanRecipientMemberIDs = nil
	delivery.HumanRecipientsTargeted = false
	delivery.Message.ChannelType = message.ConversationKindChannel

	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery))
	require.Len(t, realtime.messageNew, 1)
	require.Len(t, notifier.messages, 1, "ordinary v1 delivery keeps its historical best-effort wake")
	require.Equal(t, []string{delivery.MessageID}, marker.ids)
	require.Len(t, unread.messages, 1)
}

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

	unread := &capturingUnreadRuntime{}
	fanout := NewDeliveryFanout(nil, nil, nil).WithUnreadRuntime(unread)

	appMessage := message.AppMessage{
		ID:          "msg_followup",
		Seq:         3,
		ChannelID:   "chn_followup",
		ChannelType: "channel",
	}
	fanout.DeliverFollowupMessages(t.Context(), "wsp_1", []message.MessageDelivery{{
		MessageID:  "msg_followup",
		Seq:        3,
		AppMessage: &appMessage,
		Message: message.ReceiveMessage{
			MessageID: "msg_followup",
			TargetKey: "chn_followup",
		},
	}})

	require.Equal(t, []unreadservice.MessageStored{{
		WorkspaceID:       "wsp_1",
		ScopeID:           "chn_followup",
		ScopeType:         unreadservice.ScopeTypeChannel,
		Seq:               3,
		MessageGeneration: 3,
	}}, unread.messages)
}

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

	applyErr := errors.New("unread runtime temporarily unavailable")
	delta := unreadservice.UnreadDelta{
		WorkspaceID:  "wsp_1",
		MemberID:     "mem_reader",
		ScopeID:      "chn_first",
		Count:        1,
		HasUnread:    true,
		ServingState: unreadservice.ServingStateReady,
	}
	unread := &capturingUnreadRuntime{
		deltas:            []unreadservice.UnreadDelta{delta},
		applyMessageErrAt: 2,
		applyMessageErr:   applyErr,
	}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(nil, realtime, nil).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)

	deliveries := make([]message.MessageDelivery, 0, 2)
	for index, scopeID := range []string{"chn_first", "chn_second"} {
		seq := int64(index + 1)
		appMessage := message.AppMessage{
			ID:          fmt.Sprintf("msg_%d", seq),
			Seq:         seq,
			ChannelID:   scopeID,
			ChannelType: "channel",
		}
		deliveries = append(deliveries, message.MessageDelivery{
			MessageID:  appMessage.ID,
			Seq:        seq,
			AppMessage: &appMessage,
			Message: message.ReceiveMessage{
				MessageID:   appMessage.ID,
				TargetKey:   scopeID,
				ChannelType: "channel",
			},
		})
	}
	deliveries[0].FollowupDeliveries = append([]message.MessageDelivery(nil), deliveries[1])
	deliveries = deliveries[:1]

	fanout.DeliverFollowupMessages(t.Context(), "wsp_1", deliveries)

	require.Len(t, unread.messages, 2)
	require.Len(t, realtime.messageNew, 2, "an unreplayable ordinary unread error must not strand this delivery tree")
	require.Equal(t, []unreadservice.UnreadDelta{delta}, realtime.unreadUpdated)
}

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

	streamErr := errors.New("stream unavailable")
	stream := &fakeRealtime{appendErr: streamErr}
	notifier := &capturingNotifier{}
	unread := &capturingUnreadRuntime{}
	fanout := NewDeliveryFanout(notifier, stream, nil).WithUnreadRuntime(unread)

	appMessage := message.AppMessage{
		ID:          "msg_followup_stream_failure",
		Seq:         4,
		ChannelID:   "chn_followup",
		ChannelType: "channel",
	}
	fanout.DeliverFollowupMessages(t.Context(), "wsp_1", []message.MessageDelivery{{
		MessageID:  appMessage.ID,
		Seq:        appMessage.Seq,
		AppMessage: &appMessage,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			ChannelType: appMessage.ChannelType,
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:       "agt_followup",
			AgentMemberID: "mem_followup",
		}},
	}})

	require.Len(t, unread.messages, 1, "ordinary unread must be accounted before its non-replayable stream attempt")
	require.Empty(t, stream.unreadUpdated, "failed stream publication must not expose an unordered live delta")
	require.Empty(t, notifier.messages, "a failed follow-up stream append must not wake an agent")
}

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

	streamErr := errors.New("stream unavailable")
	stream := &fakeRealtime{appendErr: streamErr}
	unread := &capturingUnreadRuntime{}
	fanout := NewDeliveryFanout(nil, stream, nil).WithUnreadRuntime(unread)
	appMessage := message.AppMessage{
		ID:          "msg_targeted_task_followup",
		Seq:         5,
		ChannelID:   "tsk_targeted_followup",
		ChannelType: message.ConversationKindTask,
	}

	fanout.DeliverFollowupMessages(t.Context(), "wsp_1", []message.MessageDelivery{{
		MessageID:               appMessage.ID,
		Seq:                     appMessage.Seq,
		AppMessage:              &appMessage,
		HumanRecipientMemberIDs: []string{"mem_reader"},
		HumanRecipientsTargeted: true,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			ChannelType: message.ConversationKindTask,
		},
	}})

	require.Empty(t, unread.messages, "targeted Task unread replay remains safe only after its stable stream event")
	require.Empty(t, stream.unreadUpdated)
}