package task_test

import (
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

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

	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	. "github.com/tidbcloud/AgentTeam/internal/api/testkit"
	"github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/taskboard"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

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

	runtime := StartRuntime(t)
	router := gin.New()
	runtime.RegisterRoutes(router)
	server := httptest.NewServer(router)
	defer server.Close()

	client := server.Client()
	owner, workspaceID, allChannelID := SetupTaskHTTPActor(
		t,
		client,
		server.URL,
		"task-mark-read-owner@example.com",
		"task-mark-read-owner",
	)
	watcher := AcceptTaskHTTPInvite(
		t,
		runtime,
		client,
		server.URL,
		owner.User.ID,
		workspaceID,
		"task-mark-read-watcher@example.com",
		"task-mark-read-watcher",
	)
	ownerMemberID := TaskWorkspaceMemberIDForUser(t, runtime, workspaceID, owner.User.ID)
	watcherMemberID := TaskWorkspaceMemberIDForUser(t, runtime, workspaceID, watcher.User.ID)

	var created taskboard.TaskPage
	response := DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/channels/"+allChannelID+"/tasks", map[string]any{
		"tasks": []map[string]any{{"title": "Unified Task mark read"}},
	}, WorkspaceHeaders(owner.AccessToken, workspaceID), &created)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Len(t, created.Tasks, 1)
	task := created.Tasks[0]

	var policy dto.TaskNotificationPolicyResponse
	response = DoJSON(t, client, http.MethodPut, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+task.ID+"/notification-policy/self", map[string]any{
		"state": "active",
	}, WorkspaceHeaders(watcher.AccessToken, workspaceID), &policy)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Equal(t, "active", policy.Panel.Self.State)
	policyEvents, err := runtime.Store.IM().ListEventsAfter(runtime.Ctx, workspaceID, watcherMemberID, 0, 100)
	require.NoError(t, err)
	var policyEventFound bool
	for _, event := range policyEvents {
		if event.Type != "conversation.updated" || event.ScopeID != task.TaskScopeID {
			continue
		}
		if event.Payload["action"] == message.ConversationActionUnmute && event.Payload["taskId"] == task.ID {
			policyEventFound = true
		}
	}
	require.True(t, policyEventFound, "Task detail Watch must publish a member-local conversation.updated event")

	var updated dto.TaskResponse
	response = DoJSON(t, client, http.MethodPatch, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+task.ID, map[string]any{
		"description": "typed Activity for the watching member",
	}, WorkspaceHeaders(owner.AccessToken, workspaceID), &updated)
	require.Equal(t, http.StatusOK, response.StatusCode)

	var activitySeq int64
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT MAX(seq)
FROM scope_activity_events
WHERE workspace_id = $1
  AND scope_id = $2
  AND suppressed_at IS NULL
`, workspaceID, task.TaskScopeID).Scan(&activitySeq))
	require.Positive(t, activitySeq)

	_, err = runtime.UnreadRuntime.ApplyNonMessageActivity(unread.NonMessageActivityStored{
		WorkspaceID:        workspaceID,
		ScopeID:            task.TaskScopeID,
		ScopeType:          unread.ScopeTypeTask,
		Seq:                activitySeq,
		ActorMemberID:      ownerMemberID,
		ActivityGeneration: activitySeq,
		RecipientMemberIDs: []string{watcherMemberID},
	})
	require.NoError(t, err)
	require.NoError(t, runtime.Store.IM().UpsertConversation(runtime.Ctx, workspaceID, watcherMemberID, im.ConversationIndexItem{
		ConversationID: "task:" + task.TaskScopeID,
		ScopeID:        task.TaskScopeID,
		Kind:           message.ConversationKindTask,
		SortAt:         time.Now().UTC(),
		Projection: map[string]any{
			"id":      "task:" + task.TaskScopeID,
			"scopeId": task.TaskScopeID,
			"kind":    message.ConversationKindTask,
			"title":   "#t1 Unified Task mark read",
			"route":   "/channels/" + allChannelID + "/tasks/1",
		},
	}))
	require.NoError(t, message.SetConversationManualUnreadState(
		runtime.Ctx,
		runtime.Pool,
		watcherMemberID,
		task.TaskScopeID,
		0,
		time.Now().UTC(),
	))
	require.NoError(t, runtime.UnreadRuntime.ApplyManualUnread(unread.ManualUnreadChanged{
		WorkspaceID:  workspaceID,
		MemberID:     watcherMemberID,
		ScopeID:      task.TaskScopeID,
		ManualUnread: true,
	}))

	var before unread.ScopeUnreadState
	require.Eventually(t, func() bool {
		before = runtime.UnreadRuntime.ScopeStates(workspaceID, watcherMemberID, []string{task.TaskScopeID})[task.TaskScopeID]
		return before.ActivityUnreadCount == 1 && before.ManualUnread
	}, 5*time.Second, 20*time.Millisecond)
	require.Equal(t, int64(1), before.ActivityUnreadCount)
	require.True(t, before.ManualUnread)

	var lastStreamSeq int64
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT COALESCE(MAX(event_seq), 0)
FROM im_stream_events
WHERE workspace_id = $1
  AND member_id = $2
`, workspaceID, watcherMemberID).Scan(&lastStreamSeq))

	var marked dto.MarkTaskReadResponse
	response = DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+task.ID+"/read-state", nil, WorkspaceHeaders(watcher.AccessToken, workspaceID), &marked)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Zero(t, marked.ReadState.UnreadCount)
	require.False(t, marked.ReadState.ManualUnread)

	after := runtime.UnreadRuntime.ScopeStates(workspaceID, watcherMemberID, []string{task.TaskScopeID})[task.TaskScopeID]
	require.Zero(t, after.MessageUnreadCount)
	require.Zero(t, after.ActivityUnreadCount)
	require.False(t, after.ManualUnread)

	events, err := runtime.Store.IM().ListEventsAfter(runtime.Ctx, workspaceID, watcherMemberID, lastStreamSeq, 100)
	require.NoError(t, err)
	var readEventFound bool
	for _, event := range events {
		if event.Type != "conversation.read" || event.ScopeID != task.TaskScopeID {
			continue
		}
		conversation, ok := event.Payload["conversation"].(map[string]any)
		require.True(t, ok)
		require.Equal(t, float64(0), conversation["unreadCount"])
		require.NotEqual(t, true, conversation["manualUnread"])
		readEventFound = true
	}
	require.True(t, readEventFound, "Task detail mark-read must append a durable member-local conversation.read event")

	var conversations dto.ConversationPage
	response = DoJSON(t, client, http.MethodGet, server.URL+"/api/v1/workspaces/"+workspaceID+"/conversations?source=index&kind=task", nil, WorkspaceHeaders(watcher.AccessToken, workspaceID), &conversations)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Len(t, conversations.Items, 1)
	require.Equal(t, "task:"+task.TaskScopeID, conversations.Items[0].ID)
	require.Zero(t, conversations.Items[0].UnreadCount)
	require.False(t, conversations.Items[0].ManualUnread)

	var manualOnlyCreated taskboard.TaskPage
	response = DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/channels/"+allChannelID+"/tasks", map[string]any{
		"tasks": []map[string]any{{"title": "Manual-only Task mark read"}},
	}, WorkspaceHeaders(owner.AccessToken, workspaceID), &manualOnlyCreated)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Len(t, manualOnlyCreated.Tasks, 1)
	manualOnlyTask := manualOnlyCreated.Tasks[0]

	require.NoError(t, runtime.Store.IM().UpsertConversation(runtime.Ctx, workspaceID, watcherMemberID, im.ConversationIndexItem{
		ConversationID: "task:" + manualOnlyTask.TaskScopeID,
		ScopeID:        manualOnlyTask.TaskScopeID,
		Kind:           message.ConversationKindTask,
		SortAt:         time.Now().UTC(),
		Projection: map[string]any{
			"id":      "task:" + manualOnlyTask.TaskScopeID,
			"scopeId": manualOnlyTask.TaskScopeID,
			"kind":    message.ConversationKindTask,
			"title":   "#t2 Manual-only Task mark read",
			"route":   "/channels/" + allChannelID + "/tasks/2",
		},
	}))
	require.NoError(t, message.SetConversationManualUnreadState(
		runtime.Ctx,
		runtime.Pool,
		watcherMemberID,
		manualOnlyTask.TaskScopeID,
		0,
		time.Now().UTC(),
	))
	require.NoError(t, runtime.UnreadRuntime.ApplyManualUnread(unread.ManualUnreadChanged{
		WorkspaceID:  workspaceID,
		MemberID:     watcherMemberID,
		ScopeID:      manualOnlyTask.TaskScopeID,
		ManualUnread: true,
	}))

	var manualOnlyDetail dto.TaskDetailResponse
	response = DoJSON(t, client, http.MethodGet, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+manualOnlyTask.ID, nil, WorkspaceHeaders(watcher.AccessToken, workspaceID), &manualOnlyDetail)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.NotNil(t, manualOnlyDetail.Detail.Task.ReadState)
	require.Zero(t, manualOnlyDetail.Detail.Task.ReadState.UnreadCount)
	require.True(t, manualOnlyDetail.Detail.Task.ReadState.ManualUnread)

	var manualOnlyMarked dto.MarkTaskReadResponse
	response = DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+manualOnlyTask.ID+"/read-state", nil, WorkspaceHeaders(watcher.AccessToken, workspaceID), &manualOnlyMarked)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Zero(t, manualOnlyMarked.ReadState.UnreadCount)
	require.False(t, manualOnlyMarked.ReadState.ManualUnread)
}

// TestTaskDetailWatchMaterializesThenMarkReadDoesNotDuplicate proves the #t2636
// contract end to end: a human Watch (notification-policy/self active) on a Task
// immediately and idempotently materializes the owner-local Inbox conversation
// row, and a later mark-read neither removes it nor adds a second row (the
// mark-read path only publishes the durable conversation.read event).
func TestTaskDetailWatchMaterializesThenMarkReadDoesNotDuplicate(t *testing.T) {
	t.Parallel()

	runtime := StartRuntime(t)
	router := gin.New()
	runtime.RegisterRoutes(router)
	server := httptest.NewServer(router)
	defer server.Close()

	client := server.Client()
	owner, workspaceID, allChannelID := SetupTaskHTTPActor(
		t,
		client,
		server.URL,
		"task-mark-read-race-owner@example.com",
		"task-mark-read-race-owner",
	)
	watcher := AcceptTaskHTTPInvite(
		t,
		runtime,
		client,
		server.URL,
		owner.User.ID,
		workspaceID,
		"task-mark-read-race-watcher@example.com",
		"task-mark-read-race-watcher",
	)
	ownerMemberID := TaskWorkspaceMemberIDForUser(t, runtime, workspaceID, owner.User.ID)
	watcherMemberID := TaskWorkspaceMemberIDForUser(t, runtime, workspaceID, watcher.User.ID)

	var created taskboard.TaskPage
	response := DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/channels/"+allChannelID+"/tasks", map[string]any{
		"tasks": []map[string]any{{"title": "Mark read before Inbox materialization"}},
	}, WorkspaceHeaders(owner.AccessToken, workspaceID), &created)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Len(t, created.Tasks, 1)
	task := created.Tasks[0]

	var policy dto.TaskNotificationPolicyResponse
	response = DoJSON(t, client, http.MethodPut, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+task.ID+"/notification-policy/self", map[string]any{
		"state": "active",
	}, WorkspaceHeaders(watcher.AccessToken, workspaceID), &policy)
	require.Equal(t, http.StatusOK, response.StatusCode)

	var updated dto.TaskResponse
	response = DoJSON(t, client, http.MethodPatch, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+task.ID, map[string]any{
		"description": "targeted activity races the asynchronous Inbox projection",
	}, WorkspaceHeaders(owner.AccessToken, workspaceID), &updated)
	require.Equal(t, http.StatusOK, response.StatusCode)

	var activitySeq int64
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT MAX(seq)
FROM scope_activity_events
WHERE workspace_id = $1
  AND scope_id = $2
  AND suppressed_at IS NULL
`, workspaceID, task.TaskScopeID).Scan(&activitySeq))
	_, err := runtime.UnreadRuntime.ApplyNonMessageActivity(unread.NonMessageActivityStored{
		WorkspaceID:        workspaceID,
		ScopeID:            task.TaskScopeID,
		ScopeType:          unread.ScopeTypeTask,
		Seq:                activitySeq,
		ActorMemberID:      ownerMemberID,
		ActivityGeneration: activitySeq,
		RecipientMemberIDs: []string{watcherMemberID},
	})
	require.NoError(t, err)

	var indexRows int64
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT COUNT(*)
FROM im_conversation_index
WHERE workspace_id = $1
  AND member_id = $2
  AND scope_id = $3
`, workspaceID, watcherMemberID, task.TaskScopeID).Scan(&indexRows))
	require.Equal(t, int64(1), indexRows, "Task Watch must immediately materialize the owner-local Inbox conversation")

	var lastStreamSeq int64
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT COALESCE(MAX(event_seq), 0)
FROM im_stream_events
WHERE workspace_id = $1
  AND member_id = $2
`, workspaceID, watcherMemberID).Scan(&lastStreamSeq))

	var marked dto.MarkTaskReadResponse
	response = DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/tasks/"+task.ID+"/read-state", nil, WorkspaceHeaders(watcher.AccessToken, workspaceID), &marked)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.Zero(t, marked.ReadState.UnreadCount)
	require.False(t, marked.ReadState.ManualUnread)

	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT COUNT(*)
FROM im_conversation_index
WHERE workspace_id = $1
  AND member_id = $2
  AND scope_id = $3
`, workspaceID, watcherMemberID, task.TaskScopeID).Scan(&indexRows))
	require.Equal(t, int64(1), indexRows, "mark-read must not add a second Task Inbox row (Watch already materialized it)")

	events, err := runtime.Store.IM().ListEventsAfter(runtime.Ctx, workspaceID, watcherMemberID, lastStreamSeq, 100)
	require.NoError(t, err)
	var readEventFound bool
	for _, event := range events {
		if event.Type != "conversation.read" || event.ScopeID != task.TaskScopeID {
			continue
		}
		require.Equal(t, message.ConversationActionMarkRead, event.Payload["action"])
		require.Equal(t, task.ID, event.Payload["taskId"])
		// The Watch row already exists, so mark-read publishes the unified
		// conversation payload (kind task, scopeId, zero unread) rather than a
		// bare self event.
		conversation, ok := event.Payload["conversation"].(map[string]any)
		require.True(t, ok, "a materialized Task conversation must be carried on the conversation.read event")
		require.Equal(t, message.ConversationKindTask, conversation["kind"])
		require.Equal(t, task.TaskScopeID, conversation["scopeId"])
		require.Equal(t, float64(0), conversation["unreadCount"])
		readEventFound = true
	}
	require.True(t, readEventFound, "index-present mark-read must append a durable Task conversation.read event carrying the conversation")
}