package taskboard

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/coredb/testkit"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/tasktimeline"
	"github.com/tidbcloud/AgentTeam/internal/store/testfixture"
)

func TestTaskEventsProjectAndReadFromScopeActivityEvents(t *testing.T) {
	ctx := context.Background()
	fixture := seedTaskboardUserFixture(t, ctx)

	createdResponse, err := fixture.service.CreateTasks(ctx, CreateTasksRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Board:   BoardRef{ChannelID: fixture.channelID},
		Titles:  []string{"Project task activity to task scope"},
	})
	require.NoError(t, err)
	require.Len(t, createdResponse.Tasks, 1)
	task := createdResponse.Tasks[0]
	initialWorkspaceMessageSeq := loadTaskboardWorkspaceLatestMessageSeqForTest(t, ctx, fixture.pool, fixture.workspaceID)

	createdActivities := loadTaskScopeActivityEventsForTest(t, ctx, fixture.pool, fixture.workspaceID, task.TaskScopeID)
	require.Len(t, createdActivities, 1)
	require.Equal(t, int64(1), createdActivities[0].Seq)
	require.Equal(t, int64(1), createdActivities[0].WorkspaceSeq)
	require.Equal(t, createdActivities[0].Seq, createdActivities[0].ActivityGlobalSeq, "task activity events keep activity_global_seq as a scope-local compatibility alias")
	require.Equal(t, "task.created", createdActivities[0].ActivityType)
	require.Equal(t, "task_event", createdActivities[0].ActivityRefType)
	require.Equal(t, createdActivities[0].ID, createdActivities[0].ActivityRefID)
	require.Equal(t, fixture.ownerMemberID, createdActivities[0].ActorMemberID)
	require.Equal(t, "task_event", createdActivities[0].Reason)
	require.JSONEq(t, fmt.Sprintf(`{"task":{"id":%q,"scopeId":%q},"taskEvent":{"eventId":%q,"taskId":%q,"type":"created","payload":"","actorRef":%q}}`, task.ID, task.TaskScopeID, createdActivities[0].ID, task.ID, fixture.ownerMemberID), string(createdActivities[0].Payload))

	discussionResponse, err := fixture.service.SendTaskScopeMessage(ctx, SendTaskScopeMessageRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Task:    TaskRef{TaskID: task.ID},
		Message: TaskScopeMessageInput{
			Content: "Activity should be read from scope_activity_events.",
		},
	})
	require.NoError(t, err)

	activities := loadTaskScopeActivityEventsForTest(t, ctx, fixture.pool, fixture.workspaceID, task.TaskScopeID)
	require.Len(t, activities, 2)
	require.Equal(t, "task.created", activities[0].ActivityType)
	require.Equal(t, "task.discussion_added", activities[1].ActivityType)
	require.Equal(t, int64(2), activities[1].Seq)
	require.Equal(t, int64(2), activities[1].WorkspaceSeq)
	require.Equal(t, activities[1].Seq, activities[1].ActivityGlobalSeq)
	require.Equal(t, fixture.ownerMemberID, activities[1].ActorMemberID)
	require.JSONEq(t, fmt.Sprintf(`{"task":{"id":%q,"scopeId":%q},"taskEvent":{"eventId":%q,"taskId":%q,"type":"discussion_added","payload":%q,"actorRef":%q}}`, task.ID, task.TaskScopeID, activities[1].ID, task.ID, discussionResponse.Message.ID, fixture.ownerMemberID), string(activities[1].Payload))

	timelineResponse, err := fixture.service.GetTaskTimeline(ctx, GetTaskTimelineRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Task:    TaskRef{TaskID: task.ID},
	})
	require.NoError(t, err)
	require.Len(t, timelineResponse.Entries, 2)
	require.Equal(t, activities[1].ID, timelineResponse.Entries[0].ID)
	require.Equal(t, TaskTimelineEntryKindDiscussion, timelineResponse.Entries[0].Kind, "message-backed discussion activity should read through the task scope message authority")
	require.NotNil(t, timelineResponse.Entries[0].Discussion)
	require.Equal(t, discussionResponse.Message.ID, timelineResponse.Entries[0].Discussion.ID)
	require.Equal(t, "Activity should be read from scope_activity_events.", timelineResponse.Entries[0].Discussion.Body)

	var latestActivitySeq, latestMessageSeq int64
	require.NoError(t, fixture.pool.QueryRow(ctx, `
SELECT latest_activity_seq, latest_seq
FROM scopes
WHERE id = $1
`, task.TaskScopeID).Scan(&latestActivitySeq, &latestMessageSeq))
	require.Equal(t, int64(2), latestActivitySeq)
	require.Equal(t, int64(1), latestMessageSeq, "task activity events must not consume scope message seq")
	require.Equal(t, initialWorkspaceMessageSeq, loadTaskboardWorkspaceLatestMessageSeqForTest(t, ctx, fixture.pool, fixture.workspaceID), "task activity event writes must not bump the workspace message seq hot row")
}

func TestTaskTimelineReadServiceKeepsCheckpointAsTypedActivity(t *testing.T) {
	ctx := context.Background()
	fixture := seedTaskboardUserFixture(t, ctx)

	createdResponse, err := fixture.service.CreateTasks(ctx, CreateTasksRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Board:   BoardRef{ChannelID: fixture.channelID},
		Titles:  []string{"Read typed checkpoint timeline"},
	})
	require.NoError(t, err)
	require.Len(t, createdResponse.Tasks, 1)
	task := createdResponse.Tasks[0]
	checkpointText := "Checkpoint content stays in task activity payload."

	_, err = fixture.service.AddTaskCheckpoint(ctx, AddTaskCheckpointRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Task:    TaskRef{TaskID: task.ID},
		Checkpoint: TaskCheckpointInput{
			Text: checkpointText,
		},
	})
	require.NoError(t, err)

	timelineResponse, err := fixture.service.GetTaskTimeline(ctx, GetTaskTimelineRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Task:    TaskRef{TaskID: task.ID},
	})
	require.NoError(t, err)

	checkpointEvents := make([]TaskActivityEvent, 0)
	for _, event := range taskTimelineActivityEvents(timelineResponse) {
		if event.Type == taskEventCheckpointAdded {
			checkpointEvents = append(checkpointEvents, event)
		}
	}
	require.Len(t, checkpointEvents, 1)
	require.Equal(t, taskTimelineSourceKindActivity, checkpointEvents[0].SourceKind)
	require.Empty(t, checkpointEvents[0].SourceMessageID)
	require.NotEmpty(t, checkpointEvents[0].SourceActivityID)
	require.Equal(t, checkpointText, checkpointEvents[0].Payload)
	require.Equal(t, message.AuthorActorTypeHuman, checkpointEvents[0].AuthorProfile.ActorType)
	require.Equal(t, fixture.ownerUserID, checkpointEvents[0].AuthorProfile.PrincipalID)
	require.Equal(t, fixture.ownerMemberID, checkpointEvents[0].AuthorProfile.MemberID)
	require.Equal(t, fixture.ownerDisplayName, checkpointEvents[0].AuthorProfile.DisplayName)
	require.Equal(t, message.ContributionSourceKindHumanCompose, checkpointEvents[0].Source.Kind)
	require.Equal(t, message.ContributionSourceRefTypeTaskActivity, checkpointEvents[0].Source.RefType)
	require.Equal(t, checkpointEvents[0].SourceActivityID, checkpointEvents[0].Source.RefID)
	require.Equal(t, taskEventCheckpointAdded, checkpointEvents[0].Source.EventType)
	require.Equal(t, fixture.ownerMemberID, checkpointEvents[0].ActorRef)
	require.Equal(t, fixture.ownerDisplayName, checkpointEvents[0].ActorDisplayName)

	var checkpointMessageCount int
	require.NoError(t, fixture.pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE scope_id = $1
  AND content = $2
`, task.TaskScopeID, checkpointText).Scan(&checkpointMessageCount))
	require.Zero(t, checkpointMessageCount, "checkpoint must not create message-backed content that needs read-time fallback or backfill")

	timelineResponse, err = fixture.service.GetTaskTimeline(ctx, GetTaskTimelineRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Task:    TaskRef{TaskID: task.ID},
	})
	require.NoError(t, err)
	for _, entry := range taskTimelineDiscussionEntries(timelineResponse) {
		require.NotEqual(t, checkpointText, entry.Body, "checkpoint typed activity must not leak into task discussion")
	}
}

func TestTaskActivityWorkspaceSequenceRollsBackWithoutLeavingGap(t *testing.T) {
	ctx := context.Background()
	fixture := seedTaskboardUserFixture(t, ctx)

	createdResponse, err := fixture.service.CreateTasks(ctx, CreateTasksRequest{
		Context: NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Board:   BoardRef{ChannelID: fixture.channelID},
		Titles:  []string{"Rollback activity sequence"},
	})
	require.NoError(t, err)
	task := createdResponse.Tasks[0]
	require.Equal(t, int64(1), loadTaskScopeActivityEventsForTest(t, ctx, fixture.pool, fixture.workspaceID, task.TaskScopeID)[0].WorkspaceSeq)

	tx, err := fixture.pool.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	require.NoError(t, applyTaskTypedEventPlan(ctx, tx,
		taskTypedEventIntent(task.ID, fixture.ownerMemberID, taskEventUpdated, "", time.Now().UTC()),
	))
	require.NoError(t, tx.Rollback(ctx))

	_, err = fixture.service.AddTaskCheckpoint(ctx, AddTaskCheckpointRequest{
		Context:    NewUserServiceContext(fixture.workspaceID, fixture.ownerUserID, fixture.ownerDisplayName),
		Task:       TaskRef{TaskID: task.ID},
		Checkpoint: TaskCheckpointInput{Text: "Committed after rollback"},
	})
	require.NoError(t, err)

	activities := loadTaskScopeActivityEventsForTest(t, ctx, fixture.pool, fixture.workspaceID, task.TaskScopeID)
	require.Len(t, activities, 2)
	require.Equal(t, []int64{1, 2}, []int64{activities[0].WorkspaceSeq, activities[1].WorkspaceSeq})

	var lastSequence int64
	require.NoError(t, fixture.pool.QueryRow(ctx, `
SELECT last_sequence
FROM scope_activity_workspace_counters
WHERE workspace_id = $1
`, fixture.workspaceID).Scan(&lastSequence))
	require.Equal(t, int64(2), lastSequence)
}

func TestTaskActivityWorkspaceSequenceSerializesConcurrentTransactions(t *testing.T) {
	dsn := testkit.StartInitializedTiDBWithRootDSN(t, taskboardTiDB.RootDSN(t))
	// Database provisioning can be slow on a busy CI worker and should not
	// consume the deadline that guards the transaction serialization itself.
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	firstSQL, err := testkit.OpenTiDBContext(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, firstSQL.Close()) })
	firstSQL.SetMaxOpenConns(1)
	firstSQL.SetMaxIdleConns(1)
	secondSQL, err := testkit.OpenTiDBContext(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, secondSQL.Close()) })
	secondSQL.SetMaxOpenConns(1)
	secondSQL.SetMaxIdleConns(1)

	firstDB := coredb.WrapSQLDB(firstSQL)
	secondDB := coredb.WrapSQLDB(secondSQL)
	_, err = firstDB.Exec(ctx, `SET SESSION tidb_txn_mode = 'pessimistic'`)
	require.NoError(t, err)
	_, err = secondDB.Exec(ctx, `SET SESSION tidb_txn_mode = 'pessimistic'`)
	require.NoError(t, err)
	workspaceID, firstTaskID, secondTaskID := seedTaskActivitySequenceFixture(t, ctx, firstDB)
	now := time.Date(2026, 7, 13, 10, 0, 0, 0, time.UTC)

	firstTx, err := firstDB.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	firstCommitted := false
	defer func() {
		if !firstCommitted {
			_ = firstTx.Rollback(context.Background())
		}
	}()
	require.NoError(t, applyTaskTypedEventPlan(ctx, firstTx,
		tasktimeline.NewTypedActivityIntent(firstTaskID, "", taskEventUpdated, "first", now.Add(time.Hour)),
	))

	secondReachedCounter := make(chan struct{})
	secondDone := make(chan error, 1)
	go func() {
		secondTx, beginErr := secondDB.BeginTx(ctx, coredb.TxOptions{})
		if beginErr != nil {
			secondDone <- beginErr
			return
		}
		wrapped := &taskActivityCounterNotifyingTx{
			Tx:      secondTx,
			reached: secondReachedCounter,
		}
		if writeErr := applyTaskTypedEventPlan(ctx, wrapped,
			tasktimeline.NewTypedActivityIntent(secondTaskID, "", taskEventUpdated, "second", now),
		); writeErr != nil {
			_ = secondTx.Rollback(context.Background())
			secondDone <- writeErr
			return
		}
		secondDone <- secondTx.Commit(ctx)
	}()

	select {
	case <-secondReachedCounter:
	case err := <-secondDone:
		require.NoError(t, err)
		t.Fatal("second task activity transaction finished before reaching the workspace counter")
	case <-ctx.Done():
		t.Fatal("second task activity transaction did not reach the workspace counter")
	}
	select {
	case err := <-secondDone:
		require.NoError(t, err)
		t.Fatal("second task activity transaction committed before the first transaction")
	default:
	}

	require.NoError(t, firstTx.Commit(ctx))
	firstCommitted = true
	select {
	case err := <-secondDone:
		require.NoError(t, err)
	case <-ctx.Done():
		t.Fatal("second task activity transaction did not finish after the first transaction committed")
	}

	rows, err := firstDB.Query(ctx, `
SELECT ae.workspace_seq, t.id
FROM scope_activity_events ae
JOIN tasks t
  ON t.task_scope_id = ae.scope_id
WHERE ae.workspace_id = $1
  AND ae.activity_ref_type = 'task_event'
ORDER BY ae.workspace_seq
`, workspaceID)
	require.NoError(t, err)
	defer rows.Close()

	var sequences []int64
	var taskIDs []string
	for rows.Next() {
		var sequence int64
		var taskID string
		require.NoError(t, rows.Scan(&sequence, &taskID))
		sequences = append(sequences, sequence)
		taskIDs = append(taskIDs, taskID)
	}
	require.NoError(t, rows.Err())
	require.Equal(t, []int64{1, 2}, sequences)
	require.Equal(t, []string{firstTaskID, secondTaskID}, taskIDs)
}

type taskActivityCounterNotifyingTx struct {
	coredb.Tx
	once    sync.Once
	reached chan<- struct{}
}

func (tx *taskActivityCounterNotifyingTx) Exec(ctx context.Context, query string, args ...any) (coredb.Result, error) {
	if strings.Contains(query, "INSERT IGNORE INTO scope_activity_workspace_counters") {
		tx.once.Do(func() { close(tx.reached) })
	}
	return tx.Tx.Exec(ctx, query, args...)
}

func seedTaskActivitySequenceFixture(t *testing.T, ctx context.Context, db coredb.DB) (string, string, string) {
	t.Helper()

	const (
		workspaceID  = "wsp_task_activity_sequence"
		memberID     = "mem_task_activity_sequence"
		channelID    = "chn_task_activity_sequence"
		boardID      = "tb_task_activity_sequence"
		firstTaskID  = "tsk_task_activity_sequence_first"
		secondTaskID = "tsk_task_activity_sequence_second"
	)
	now := time.Date(2026, 7, 13, 9, 0, 0, 0, time.UTC)
	organizationID := testfixture.SeedWorkspaceOrganization(t, ctx, db, workspaceID, now)
	ownerUserID := "usr_owner_" + workspaceID

	mustExecTaskboard(t, ctx, db, `
INSERT INTO workspaces (id, organization_id, slug, name, latest_message_seq, created_at, updated_at)
VALUES ($1, $2, 'task-activity-sequence', 'Task Activity Sequence', 0, $3, $3)
`, workspaceID, organizationID, now)
	mustExecTaskboard(t, ctx, db, `
INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, status, created_at, updated_at)
VALUES ($1, $2, 'human', 'Task Activity Sequence Owner', $3, 'owner', 'active', $4, $4)
`, memberID, workspaceID, ownerUserID, now)
	mustExecTaskboard(t, ctx, db, `
INSERT INTO scopes (
  id, workspace_id, type, identity_key, latest_seq, latest_global_seq,
  latest_message_at, status, created_by_member_id, created_at, updated_at
) VALUES ($1, $2, 'channel', 'channel:task-activity-sequence', 0, 0, NULL, 'active', $3, $4, $4)
`, channelID, workspaceID, memberID, now)
	mustExecTaskboard(t, ctx, db, `
INSERT INTO task_boards (
  id, channel_id, workspace_id, next_task_number, created_at, updated_at
) VALUES ($1, $2, $3, 3, $4, $4)
`, boardID, channelID, workspaceID, now)
	mustExecTaskboard(t, ctx, db, `
INSERT INTO scopes (
  id, workspace_id, type, identity_key, latest_seq, latest_global_seq,
  latest_message_at, status, created_by_member_id, created_at, updated_at
) VALUES
  ($1, $3, 'task', $4, 0, 0, NULL, 'active', $5, $6, $6),
  ($2, $3, 'task', $7, 0, 0, NULL, 'active', $5, $6, $6)
`, firstTaskID, secondTaskID, workspaceID, "task:"+firstTaskID, memberID, now, "task:"+secondTaskID)
	mustExecTaskboard(t, ctx, db, `
INSERT INTO tasks (
  id, task_scope_id, board_id, channel_id, task_number, title, status,
  created_by_member_id, created_by_ref, created_at, updated_at
) VALUES
  ($1, $1, $3, $4, 1, 'First concurrent activity', 'todo', $5, $5, $6, $6),
  ($2, $2, $3, $4, 2, 'Second concurrent activity', 'todo', $5, $5, $6, $6)
`, firstTaskID, secondTaskID, boardID, channelID, memberID, now)
	mustExecTaskboard(t, ctx, db, `
INSERT INTO scope_activity_workspace_counters (workspace_id, last_sequence, updated_at)
VALUES ($1, 0, $2)
`, workspaceID, now)

	return workspaceID, firstTaskID, secondTaskID
}

type taskScopeActivityEventForTest struct {
	ID                string
	Seq               int64
	ActivityGlobalSeq int64
	WorkspaceSeq      int64
	ActivityType      string
	ActivityRefType   string
	ActivityRefID     string
	ActorMemberID     string
	Reason            string
	Payload           []byte
}

func loadTaskScopeActivityEventsForTest(t *testing.T, ctx context.Context, db coredb.Queryer, workspaceID string, taskScopeID string) []taskScopeActivityEventForTest {
	t.Helper()

	rows, err := db.Query(ctx, `
SELECT id, seq, activity_global_seq, workspace_seq, activity_type, activity_ref_type, activity_ref_id,
       CASE WHEN actor_member_id IS NULL THEN '' ELSE actor_member_id END, reason, payload
FROM scope_activity_events
WHERE workspace_id = $1
  AND scope_id = $2
  AND activity_ref_type = 'task_event'
ORDER BY seq ASC
`, workspaceID, taskScopeID)
	require.NoError(t, err)
	defer rows.Close()

	events := make([]taskScopeActivityEventForTest, 0)
	for rows.Next() {
		var event taskScopeActivityEventForTest
		require.NoError(t, rows.Scan(
			&event.ID,
			&event.Seq,
			&event.ActivityGlobalSeq,
			&event.WorkspaceSeq,
			&event.ActivityType,
			&event.ActivityRefType,
			&event.ActivityRefID,
			&event.ActorMemberID,
			&event.Reason,
			&event.Payload,
		))
		events = append(events, event)
	}
	require.NoError(t, rows.Err())
	return events
}