package fanout

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

type capturingRecallStreamWriter struct {
	requests []im.ScopeEventRequest
}

func (writer *capturingRecallStreamWriter) AppendScopeEvent(_ context.Context, request im.ScopeEventRequest) ([]im.StreamEvent, error) {
	writer.requests = append(writer.requests, request)
	return nil, nil
}

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

	writer := &capturingRecallStreamWriter{}
	fanout := NewDeliveryFanout(nil, writer, nil)
	appMessage := message.AppMessage{
		ID:          "msg_notification_recalled",
		Seq:         19,
		ChannelID:   "chn_notifications",
		ChannelType: "channel",
		Content:     "通知已撤回",
		CreatedAt:   time.Date(2026, 7, 10, 8, 0, 0, 0, time.UTC),
		Metadata: map[string]any{
			message.MessageRecallMetadataKey: map[string]any{
				"status": "recalled",
				"kind":   "service_notification_withdrawal",
			},
		},
	}

	fanout.EmitMessageRecalled(t.Context(), "wsp_1", appMessage, true)

	require.Len(t, writer.requests, 2)
	request := writer.requests[0]
	require.Equal(t, "wsp_1", request.WorkspaceID)
	require.Equal(t, "chn_notifications", request.ScopeID)
	require.Equal(t, imEventTypeMessageRecalled, request.Event.Type)
	require.Equal(t, "msg_notification_recalled", request.Event.EntityID)
	require.NotNil(t, request.Event.EntitySeq)
	require.EqualValues(t, 19, *request.Event.EntitySeq)
	require.Equal(t, "chn_notifications", request.Event.Payload["scopeId"])
	require.Equal(t, "msg_notification_recalled", request.Event.Payload["messageId"])
	require.EqualValues(t, 19, request.Event.Payload["seq"])
	require.NotNil(t, request.Event.Payload["message"])
	require.Equal(t, appMessage.Metadata[message.MessageRecallMetadataKey], request.Event.Payload["tombstone"])
	require.False(t, request.Event.CreatedAt.IsZero())
	require.NotNil(t, request.Conversation)
	require.Equal(t, "chn_notifications", request.Conversation.ScopeID)
	require.NotNil(t, request.Conversation.Projection["lastMessage"])
	pins := writer.requests[1]
	require.Equal(t, imEventTypeMessagePinsInvalidated, pins.Event.Type)
	require.Equal(t, "chn_notifications", pins.ScopeID)
	require.Equal(t, []string{"msg_notification_recalled"}, pins.Event.Payload["messageIds"])
}

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

	writer := &capturingRecallStreamWriter{}
	fanout := NewDeliveryFanout(nil, writer, nil)

	fanout.EmitMessageRecalled(t.Context(), "wsp_1", message.AppMessage{
		ID:        "msg_notification_recalled_older",
		Seq:       7,
		ChannelID: "chn_notifications",
		Content:   "通知已撤回",
	}, false)

	require.Len(t, writer.requests, 2)
	require.Nil(t, writer.requests[0].Conversation)
}

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

	writer := &capturingRecallStreamWriter{}
	value := NewDeliveryFanout(nil, writer, nil)
	value.EmitMessageRecalled(t.Context(), "wsp_1", message.AppMessage{
		ID: "msg_thread_recalled", Seq: 8, ChannelID: "thr_1", ChannelType: "thread", ParentChannelID: "chn_parent",
		Content: "通知已撤回",
	}, false)

	require.Len(t, writer.requests, 2)
	require.Equal(t, imEventTypeMessageRecalled, writer.requests[0].Event.Type)
	require.Equal(t, "thr_1", writer.requests[1].ScopeID)
	require.Equal(t, imEventTypeMessagePinsInvalidated, writer.requests[1].Event.Type)
}

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

	updated := message.AppMessage{
		ID: "msg_notification_updated", Seq: 20, ChannelID: "chn_notifications",
		Content: "已审批", EditRevision: 2,
	}
	updatePayload, err := message.MarshalScopeMessageMutationFanoutPayload(message.ScopeMessageMutationFanoutPayload{
		Kind: message.ScopeMessageMutationUpdate, Message: updated,
	})
	require.NoError(t, err)
	writer := &capturingRecallStreamWriter{}
	adapter := NewScopeFanoutReplayAdapter(NewDeliveryFanout(nil, writer, nil))
	updateRecord := message.ScopeFanoutOutboxRecord{
		WorkspaceID: "wsp_1", ScopeID: updated.ChannelID, MessageID: updated.ID,
		EventRefType: message.ScopeMessageMutationEventRefType, EventRefID: "btnm_update_1",
		FanoutType: message.ScopeOutboxFanoutMessageMutation, Payload: updatePayload,
	}

	require.NoError(t, adapter.Fanout(t.Context(), updateRecord))
	require.Len(t, writer.requests, 1)
	require.Equal(t, imEventTypeMessageUpdated, writer.requests[0].Event.Type)
	require.Equal(t, "message.updated:btnm_update_1:chn_notifications", writer.requests[0].Event.EventID)
	require.EqualValues(t, 2, writer.requests[0].Event.Payload["editRevision"])

	recalled := updated
	recalled.Metadata = map[string]any{
		message.MessageRecallMetadataKey: map[string]any{"status": "recalled"},
	}
	recallPayload, err := message.MarshalScopeMessageMutationFanoutPayload(message.ScopeMessageMutationFanoutPayload{
		Kind: message.ScopeMessageMutationRecall, Message: recalled, UpdateConversation: true,
	})
	require.NoError(t, err)
	recallRecord := updateRecord
	recallRecord.EventRefID = "btnm_recall_1"
	recallRecord.Payload = recallPayload

	require.NoError(t, adapter.Fanout(t.Context(), recallRecord))
	require.Len(t, writer.requests, 3)
	require.Equal(t, "message.recalled:btnm_recall_1:chn_notifications", writer.requests[1].Event.EventID)
	require.Equal(t, "message.pins.invalidated:btnm_recall_1:chn_notifications", writer.requests[2].Event.EventID)
}