package im

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/task"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

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

	lastActivityAt := time.Date(2026, 7, 29, 1, 0, 0, 0, time.UTC)
	streamWriter := &recordingStreamWriter{}
	unread := &fakeUnreadRuntime{
		states: map[string]unread.ScopeUnreadState{
			"task_scope_1": {
				ScopeType:           unread.ScopeTypeTask,
				MessageUnreadCount:  2,
				ActivityUnreadCount: 3,
				ManualUnread:        true,
				ServingState:        unread.ServingStateReady,
			},
		},
	}
	sink := &recordingUnreadSink{}
	projector := NewTaskConversationProjector(
		fakeConversationStore{item: message.ConversationItem{
			Kind:           message.ConversationKindTask,
			ScopeID:        "task_scope_1",
			Title:          "#t1 Project read state",
			LastActivityAt: &lastActivityAt,
			UnreadCount:    9,
			ManualUnread:   true,
			Task: &message.ConversationTaskMetadata{
				ID:      "tsk_1",
				ScopeID: "task_scope_1",
				Title:   "Project read state",
			},
		}},
		streamWriter,
		unread,
		sink,
	)

	err := projector.ProjectTaskConversationRead(context.Background(), task.TaskConversationReadProjection{
		WorkspaceID:         "wsp_1",
		UserID:              "usr_1",
		MemberID:            "mem_1",
		TaskID:              "tsk_1",
		ScopeID:             "task_scope_1",
		LastMessageReadSeq:  12,
		LastActivityReadSeq: 34,
	})
	require.NoError(t, err)

	require.Len(t, sink.unreadUpdated, 1)
	require.Equal(t, int64(5), sink.unreadUpdated[0].Count)
	require.True(t, sink.unreadUpdated[0].ManualUnread)
	require.True(t, sink.unreadUpdated[0].HasUnread)

	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, "task_scope_1", recorded.scopeID)
	require.Equal(t, imEventTypeConversationRead, recorded.event.Type)
	require.Equal(t, "tsk_1", recorded.event.Payload["taskId"])
	require.NotNil(t, recorded.conversation)
	require.Equal(t, "task:task_scope_1", recorded.conversation.ConversationID)
	require.Equal(t, message.ConversationKindTask, recorded.conversation.Kind)

	payloadConversation, ok := recorded.event.Payload["conversation"].(dto.ConversationItem)
	require.True(t, ok)
	require.Equal(t, int64(5), payloadConversation.UnreadCount)
	require.True(t, payloadConversation.ManualUnread)
}

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

	streamWriter := &recordingStreamWriter{}
	unread := &fakeUnreadRuntime{
		states: map[string]unread.ScopeUnreadState{
			"task_scope_1": {
				ScopeType:           unread.ScopeTypeTask,
				MessageUnreadCount:  1,
				ActivityUnreadCount: 2,
				ManualUnread:        true,
				ServingState:        unread.ServingStateReady,
			},
		},
	}
	sink := &recordingUnreadSink{}
	projector := NewTaskConversationProjector(
		fakeConversationStore{loadErr: message.ErrNotFound},
		streamWriter,
		unread,
		sink,
	)

	err := projector.ProjectTaskConversationRead(context.Background(), task.TaskConversationReadProjection{
		WorkspaceID:         "wsp_1",
		UserID:              "usr_1",
		MemberID:            "mem_1",
		TaskID:              "tsk_1",
		ScopeID:             "task_scope_1",
		LastMessageReadSeq:  12,
		LastActivityReadSeq: 34,
	})
	require.NoError(t, err)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, imEventTypeConversationRead, recorded.event.Type)
	require.Equal(t, message.ConversationActionMarkRead, recorded.event.Payload["action"])
	require.Equal(t, message.ConversationKindTask, recorded.event.Payload["kind"])
	require.Equal(t, "tsk_1", recorded.event.Payload["taskId"])
	require.Equal(t, "task_scope_1", recorded.event.Payload["scopeId"])
	require.Equal(t, int64(3), recorded.event.Payload["unreadCount"])
	require.Equal(t, true, recorded.event.Payload["manualUnread"])
	require.Nil(t, recorded.conversation, "mark-read must not create an Inbox row before attention materializes one")
	require.Len(t, sink.unreadUpdated, 1)
	require.Equal(t, int64(3), sink.unreadUpdated[0].Count)
	require.True(t, sink.unreadUpdated[0].ManualUnread)
}

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

	streamWriter := &recordingStreamWriter{}
	projector := NewTaskConversationProjector(
		fakeConversationStore{loadErr: errors.New("database unavailable")},
		streamWriter,
		nil,
		nil,
	)

	err := projector.ProjectTaskConversationRead(context.Background(), task.TaskConversationReadProjection{
		WorkspaceID: "wsp_1",
		UserID:      "usr_1",
		MemberID:    "mem_1",
		TaskID:      "tsk_1",
		ScopeID:     "task_scope_1",
	})

	require.ErrorContains(t, err, "load Task conversation after mark read")
	require.Empty(t, streamWriter.memberRequests)
}

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

	streamWriter := &recordingStreamWriter{}
	projector := NewTaskConversationProjector(
		fakeConversationStore{item: message.ConversationItem{
			Kind:              message.ConversationKindTask,
			ScopeID:           "task_scope_1",
			SubscriptionState: message.HumanScopeSubscriptionStateMute,
			Task: &message.ConversationTaskMetadata{
				ID:      "tsk_1",
				ScopeID: "task_scope_1",
				Title:   "Watch parity",
			},
		}},
		streamWriter,
		nil,
		nil,
	)

	err := projector.ProjectTaskConversationPolicy(context.Background(), task.TaskConversationPolicyProjection{
		WorkspaceID: "wsp_1",
		UserID:      "usr_1",
		MemberID:    "mem_1",
		TaskID:      "tsk_1",
		ScopeID:     "task_scope_1",
		State:       message.HumanScopeSubscriptionStateActive,
	})
	require.NoError(t, err)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, imEventTypeConversationUpdated, recorded.event.Type)
	require.Equal(t, message.ConversationActionUnmute, recorded.event.Payload["action"])
	require.Equal(t, "tsk_1", recorded.event.Payload["taskId"])
	require.NotNil(t, recorded.conversation)
	payloadConversation, ok := recorded.event.Payload["conversation"].(dto.ConversationItem)
	require.True(t, ok)
	require.Equal(t, message.HumanScopeSubscriptionStateActive, payloadConversation.SubscriptionState)
}

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

	streamWriter := &recordingStreamWriter{}
	projector := NewTaskConversationProjector(
		fakeConversationStore{loadErr: message.ErrNotFound},
		streamWriter,
		nil,
		nil,
	)

	err := projector.ProjectTaskConversationPolicy(context.Background(), task.TaskConversationPolicyProjection{
		WorkspaceID: "wsp_1",
		UserID:      "usr_1",
		MemberID:    "mem_1",
		TaskID:      "tsk_1",
		ScopeID:     "task_scope_1",
		State:       message.HumanScopeSubscriptionStateMute,
	})
	require.NoError(t, err)
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, imEventTypeConversationUpdated, recorded.event.Type)
	require.Equal(t, message.ConversationActionMute, recorded.event.Payload["action"])
	require.Equal(t, message.ConversationKindTask, recorded.event.Payload["kind"])
	require.Equal(t, "tsk_1", recorded.event.Payload["taskId"])
	require.Equal(t, "task_scope_1", recorded.event.Payload["scopeId"])
	require.Nil(t, recorded.conversation, "Mute alone must not create an Inbox candidate")
}

// TestTaskConversationPolicyProjectorMaterializesOnWatchWhenIndexMissing proves
// the post-commit fallback: a Watch (active) on a Task with no owner-local
// conversation row yet must materialize it idempotently through the canonical
// materializer, so shared/Web/miniapp realtime see the Task immediately instead
// of waiting for a later targeted event.
func TestTaskConversationPolicyProjectorMaterializesOnWatchWhenIndexMissing(t *testing.T) {
	t.Parallel()

	streamWriter := &recordingStreamWriter{}
	store := &materializingConversationStore{
		fakeConversationStore: fakeConversationStore{loadErr: message.ErrNotFound},
	}
	projector := NewTaskConversationProjector(store, streamWriter, nil, nil)

	err := projector.ProjectTaskConversationPolicy(context.Background(), task.TaskConversationPolicyProjection{
		WorkspaceID: "wsp_1",
		UserID:      "usr_1",
		MemberID:    "mem_1",
		TaskID:      "tsk_1",
		ScopeID:     "task_scope_1",
		State:       message.HumanScopeSubscriptionStateActive,
	})
	require.NoError(t, err)
	require.Equal(t, 1, store.materializeCalls, "Watch on a missing Inbox row must materialize the owner-local conversation")
	require.Len(t, streamWriter.memberRequests, 1)
	recorded := streamWriter.memberRequests[0]
	require.Equal(t, imEventTypeConversationUpdated, recorded.event.Type)
	require.Equal(t, message.ConversationActionUnmute, recorded.event.Payload["action"])
	require.Equal(t, message.ConversationKindTask, recorded.event.Payload["kind"])
	require.Equal(t, "tsk_1", recorded.event.Payload["taskId"])
	require.Equal(t, "task_scope_1", recorded.event.Payload["scopeId"])
}

// materializingConversationStore is a fakeConversationStore that also implements
// taskConversationMaterializer so the projector can materialize on Watch.
type materializingConversationStore struct {
	fakeConversationStore
	materializeCalls int
	materializeErr   error
}

func (store *materializingConversationStore) MaterializeTaskConversationForMember(context.Context, string, string, string) error {
	store.materializeCalls++
	return store.materializeErr
}

type taskConversationLifecycleStoreStub struct {
	ConversationStore
	active        bool
	calls         int
	callbackCalls int
	workspaceID   string
	scopeIDs      []string
}

func (store *taskConversationLifecycleStoreStub) WithActiveTaskScopes(
	_ context.Context,
	workspaceID string,
	scopeIDs []string,
	project func() error,
) (bool, error) {
	store.calls++
	store.workspaceID = workspaceID
	store.scopeIDs = append([]string(nil), scopeIDs...)
	if !store.active {
		return false, nil
	}
	store.callbackCalls++
	return true, project()
}

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

	for _, project := range []struct {
		name string
		run  func(*TaskConversationProjector) error
	}{
		{
			name: "read",
			run: func(projector *TaskConversationProjector) error {
				return projector.ProjectTaskConversationRead(t.Context(), task.TaskConversationReadProjection{
					WorkspaceID: "wsp_1",
					UserID:      "usr_1",
					MemberID:    "mem_1",
					TaskID:      "tsk_1",
					ScopeID:     "task_scope_1",
				})
			},
		},
		{
			name: "policy",
			run: func(projector *TaskConversationProjector) error {
				return projector.ProjectTaskConversationPolicy(t.Context(), task.TaskConversationPolicyProjection{
					WorkspaceID: "wsp_1",
					UserID:      "usr_1",
					MemberID:    "mem_1",
					TaskID:      "tsk_1",
					ScopeID:     "task_scope_1",
					State:       message.HumanScopeSubscriptionStateActive,
				})
			},
		},
	} {
		project := project
		t.Run(project.name, func(t *testing.T) {
			t.Parallel()

			store := &taskConversationLifecycleStoreStub{
				ConversationStore: fakeConversationStore{loadErr: message.ErrNotFound},
			}
			streamWriter := &recordingStreamWriter{}
			sink := &recordingUnreadSink{}
			projector := NewTaskConversationProjector(store, streamWriter, nil, sink)

			require.NoError(t, project.run(projector))
			require.Equal(t, 1, store.calls)
			require.Zero(t, store.callbackCalls)
			require.Equal(t, "wsp_1", store.workspaceID)
			require.Equal(t, []string{"task_scope_1"}, store.scopeIDs)
			require.Empty(t, streamWriter.memberRequests)
			require.Empty(t, sink.unreadUpdated)
		})
	}
}