package taskboard

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

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

type concurrentDeleteTaskResult struct {
	response DeleteTaskResponse
	err      error
}

type concurrentUpdateTaskResult struct {
	response UpdateTaskResponse
	err      error
}

func TestDeleteTaskSerializesWithConcurrentSubtaskCreation(t *testing.T) {
	tests := []struct {
		name         string
		serviceCtx   func(taskboardUserFixture) ServiceContext
		subtaskMode  DeleteTaskSubtaskMode
		wantPromoted bool
	}{
		{
			name: "human create and promote",
			serviceCtx: func(fixture taskboardUserFixture) ServiceContext {
				return taskboardUserContext(fixture)
			},
			subtaskMode:  DeleteTaskSubtaskModePromote,
			wantPromoted: true,
		},
		{
			name: "agent create and recursive delete",
			serviceCtx: func(fixture taskboardUserFixture) ServiceContext {
				return NewAgentServiceContext(fixture.workspaceID, fixture.agentID, fixture.agentMemberID)
			},
			subtaskMode: DeleteTaskSubtaskModeDelete,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
			defer cancel()
			fixture := seedTaskboardUserFixtureOnDB(t, ctx, startTaskReadLifecycleTiDB(t, ctx))
			sink := &recordingTaskChangeSink{}
			fixture.service.taskChangeSink = sink
			serviceCtx := test.serviceCtx(fixture)

			parentResponse, err := fixture.service.CreateTasks(ctx, CreateTasksRequest{
				Context: serviceCtx,
				Board:   taskboardBoardRefForContext(serviceCtx, fixture.channelID),
				Titles:  []string{"Concurrent delete parent"},
			})
			require.NoError(t, err)
			require.Len(t, parentResponse.Tasks, 1)
			parent := parentResponse.Tasks[0]
			sink.events = nil

			createTx, err := fixture.pool.BeginTx(ctx, coredb.TxOptions{})
			require.NoError(t, err)
			createCommitted := false
			defer func() {
				if !createCommitted {
					_ = createTx.Rollback(context.Background())
				}
			}()

			parentRef := TaskRef{TaskID: parent.ID}
			if serviceCtx.principalType() == PrincipalTypeAgent {
				parentRef = TaskRef{TaskNumber: parent.TaskNumber}
			}
			childResponse, err := fixture.service.InTx(createTx).CreateSubtasks(ctx, CreateSubtasksRequest{
				Context: serviceCtx,
				Board:   taskboardBoardRefForContext(serviceCtx, fixture.channelID),
				Task:    parentRef,
				Titles:  []string{"Concurrent child"},
			})
			require.NoError(t, err)
			require.Len(t, childResponse.Tasks, 1)
			child := childResponse.Tasks[0]

			deleteDone := make(chan concurrentDeleteTaskResult, 1)
			go func() {
				response, deleteErr := fixture.service.DeleteTask(ctx, DeleteTaskRequest{
					Context:     serviceCtx,
					Board:       taskboardBoardRefForContext(serviceCtx, fixture.channelID),
					Task:        parentRef,
					SubtaskMode: test.subtaskMode,
				})
				deleteDone <- concurrentDeleteTaskResult{response: response, err: deleteErr}
			}()

			requireConcurrentDeleteBlocked(t, deleteDone)
			require.NoError(t, createTx.Commit(ctx))
			createCommitted = true

			result := requireConcurrentDeleteDone(t, deleteDone)
			require.NoError(t, result.err)
			require.Zero(t, countTaskRowsByID(t, ctx, fixture, parent.ID))
			require.Equal(t, "deleted", taskScopeStatus(t, ctx, fixture, parent.TaskScopeID))
			require.Zero(t, countDanglingTaskParentRefs(t, ctx, fixture))

			if test.wantPromoted {
				require.Equal(t, []int64{parent.TaskNumber}, result.response.DeletedTaskNumbers)
				require.Equal(t, []int64{child.TaskNumber}, result.response.PromotedTaskNumbers)
				require.Equal(t, int64(1), countTaskRowsByID(t, ctx, fixture, child.ID))
				require.Empty(t, taskParentRef(t, ctx, fixture, child.ID))
				require.Equal(t, "active", taskScopeStatus(t, ctx, fixture, child.TaskScopeID))
				require.Len(t, sink.events, 2)
				require.Equal(t, []TaskChangeType{TaskChangeUpdated, TaskChangeDeleted}, []TaskChangeType{
					sink.events[0].EventType,
					sink.events[1].EventType,
				})
				require.Equal(t, child.ID, sink.events[0].Task.ID)
				require.Equal(t, parent.ID, sink.events[1].TaskID)
				return
			}

			require.Equal(t, []int64{child.TaskNumber, parent.TaskNumber}, result.response.DeletedTaskNumbers)
			require.Empty(t, result.response.PromotedTaskNumbers)
			require.Zero(t, countTaskRowsByID(t, ctx, fixture, child.ID))
			require.Equal(t, "deleted", taskScopeStatus(t, ctx, fixture, child.TaskScopeID))
			require.Len(t, sink.events, 2)
			require.Equal(t, []TaskChangeType{TaskChangeDeleted, TaskChangeDeleted}, []TaskChangeType{
				sink.events[0].EventType,
				sink.events[1].EventType,
			})
			require.Equal(t, child.ID, sink.events[0].TaskID)
			require.Equal(t, parent.ID, sink.events[1].TaskID)
		})
	}
}

func TestDeleteTaskSerializesWithConcurrentParentAttachment(t *testing.T) {
	tests := []struct {
		name        string
		serviceCtx  func(taskboardUserFixture) ServiceContext
		subtaskMode DeleteTaskSubtaskMode
	}{
		{
			name: "human attach and promote",
			serviceCtx: func(fixture taskboardUserFixture) ServiceContext {
				return taskboardUserContext(fixture)
			},
			subtaskMode: DeleteTaskSubtaskModePromote,
		},
		{
			name: "agent attach and recursive delete",
			serviceCtx: func(fixture taskboardUserFixture) ServiceContext {
				return NewAgentServiceContext(fixture.workspaceID, fixture.agentID, fixture.agentMemberID)
			},
			subtaskMode: DeleteTaskSubtaskModeDelete,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
			defer cancel()
			fixture := seedTaskboardUserFixtureOnDB(t, ctx, startTaskReadLifecycleTiDB(t, ctx))
			serviceCtx := test.serviceCtx(fixture)
			board := taskboardBoardRefForContext(serviceCtx, fixture.channelID)
			created, err := fixture.service.CreateTasks(ctx, CreateTasksRequest{
				Context: serviceCtx,
				Board:   board,
				Titles:  []string{"Deleting parent", "Root task to attach"},
			})
			require.NoError(t, err)
			require.Len(t, created.Tasks, 2)
			parent := created.Tasks[0]
			child := created.Tasks[1]

			sink := &recordingTaskChangeSink{}
			barrierDB := newTaskDeleteCommitBarrierDB(fixture.pool)
			fixture.service = NewWithOptions(barrierDB, StoreOptions{TaskChangeSink: sink})

			deleteDone := make(chan concurrentDeleteTaskResult, 1)
			go func() {
				response, deleteErr := fixture.service.DeleteTask(ctx, DeleteTaskRequest{
					Context:     serviceCtx,
					Board:       board,
					Task:        taskRefForServiceContext(serviceCtx, parent),
					SubtaskMode: test.subtaskMode,
				})
				deleteDone <- concurrentDeleteTaskResult{response: response, err: deleteErr}
			}()

			requireTaskDeleteReachedCommit(t, ctx, barrierDB)

			updateDone := make(chan concurrentUpdateTaskResult, 1)
			go func() {
				response, updateErr := fixture.service.UpdateTask(ctx, UpdateTaskRequest{
					Context: serviceCtx,
					Board:   board,
					Task:    taskRefForServiceContext(serviceCtx, child),
					Patch: TaskUpdatePatch{
						ParentTaskRef: stringPointer(parent.ID),
						AssigneeRef:   stringPointer(fixture.agentMemberID),
					},
				})
				updateDone <- concurrentUpdateTaskResult{response: response, err: updateErr}
			}()

			var earlyUpdate *concurrentUpdateTaskResult
			select {
			case result := <-updateDone:
				earlyUpdate = &result
			case <-time.After(250 * time.Millisecond):
			}
			barrierDB.releaseDeleteCommit()
			if earlyUpdate != nil {
				t.Fatalf("parent attachment completed while delete was waiting to commit: %v", earlyUpdate.err)
			}

			deleteResult := requireConcurrentDeleteDone(t, deleteDone)
			require.NoError(t, deleteResult.err)
			require.Equal(t, []int64{parent.TaskNumber}, deleteResult.response.DeletedTaskNumbers)
			require.Empty(t, deleteResult.response.PromotedTaskNumbers)

			updateResult := requireConcurrentUpdateDone(t, updateDone)
			require.Error(t, updateResult.err)
			require.True(t,
				errors.Is(updateResult.err, ErrNotFound) || errors.Is(updateResult.err, ErrBadRequest),
				"parent attachment must fail after delete, got %v",
				updateResult.err,
			)
			require.Zero(t, countTaskRowsByID(t, ctx, fixture, parent.ID))
			require.Equal(t, "deleted", taskScopeStatus(t, ctx, fixture, parent.TaskScopeID))
			require.Equal(t, int64(1), countTaskRowsByID(t, ctx, fixture, child.ID))
			require.Empty(t, taskParentRef(t, ctx, fixture, child.ID))
			require.Equal(t, "active", taskScopeStatus(t, ctx, fixture, child.TaskScopeID))
			require.Zero(t, countDanglingTaskParentRefs(t, ctx, fixture))
			require.Len(t, sink.events, 1)
			require.Equal(t, TaskChangeDeleted, sink.events[0].EventType)
			require.Equal(t, parent.ID, sink.events[0].TaskID)
		})
	}
}

type taskDeleteCommitBarrierDB struct {
	coredb.DB
	mu              sync.Mutex
	barrierAssigned bool
	commitReached   chan struct{}
	commitRelease   chan struct{}
	releaseOnce     sync.Once
}

func newTaskDeleteCommitBarrierDB(db coredb.DB) *taskDeleteCommitBarrierDB {
	return &taskDeleteCommitBarrierDB{
		DB:            db,
		commitReached: make(chan struct{}),
		commitRelease: make(chan struct{}),
	}
}

func (db *taskDeleteCommitBarrierDB) BeginTx(ctx context.Context, options coredb.TxOptions) (coredb.Tx, error) {
	tx, err := db.DB.BeginTx(ctx, options)
	if err != nil {
		return nil, err
	}
	db.mu.Lock()
	useBarrier := !db.barrierAssigned
	if useBarrier {
		db.barrierAssigned = true
	}
	db.mu.Unlock()
	if !useBarrier {
		return tx, nil
	}
	return &taskDeleteCommitBarrierTx{
		Tx:      tx,
		reached: db.commitReached,
		release: db.commitRelease,
	}, nil
}

func (db *taskDeleteCommitBarrierDB) releaseDeleteCommit() {
	db.releaseOnce.Do(func() {
		close(db.commitRelease)
	})
}

type taskDeleteCommitBarrierTx struct {
	coredb.Tx
	reached chan struct{}
	release chan struct{}
}

func (tx *taskDeleteCommitBarrierTx) Commit(ctx context.Context) error {
	close(tx.reached)
	select {
	case <-tx.release:
		return tx.Tx.Commit(ctx)
	case <-ctx.Done():
		return ctx.Err()
	}
}

func requireTaskDeleteReachedCommit(t *testing.T, ctx context.Context, db *taskDeleteCommitBarrierDB) {
	t.Helper()
	select {
	case <-db.commitReached:
	case <-ctx.Done():
		t.Fatal("delete did not reach its commit barrier")
	}
}

func taskRefForServiceContext(serviceCtx ServiceContext, task Task) TaskRef {
	if serviceCtx.principalType() == PrincipalTypeAgent {
		return TaskRef{TaskNumber: task.TaskNumber}
	}
	return TaskRef{TaskID: task.ID}
}

func requireConcurrentUpdateDone(t *testing.T, done <-chan concurrentUpdateTaskResult) concurrentUpdateTaskResult {
	t.Helper()
	select {
	case result := <-done:
		return result
	case <-time.After(10 * time.Second):
		t.Fatal("timed out waiting for concurrent Task update")
		return concurrentUpdateTaskResult{}
	}
}

func requireConcurrentDeleteBlocked(t *testing.T, done <-chan concurrentDeleteTaskResult) {
	t.Helper()
	select {
	case result := <-done:
		t.Fatalf("delete completed before the concurrent subtask committed: %v", result.err)
	case <-time.After(250 * time.Millisecond):
	}
}

func requireConcurrentDeleteDone(t *testing.T, done <-chan concurrentDeleteTaskResult) concurrentDeleteTaskResult {
	t.Helper()
	select {
	case result := <-done:
		return result
	case <-time.After(10 * time.Second):
		t.Fatal("timed out waiting for concurrent Task delete")
		return concurrentDeleteTaskResult{}
	}
}

func countDanglingTaskParentRefs(t *testing.T, ctx context.Context, fixture taskboardUserFixture) int64 {
	t.Helper()
	var count int64
	require.NoError(t, fixture.pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM tasks child
JOIN task_boards tb ON tb.id = child.board_id
LEFT JOIN tasks parent ON parent.id = child.parent_task_ref
WHERE tb.workspace_id = $1
  AND child.parent_task_ref <> ''
  AND parent.id IS NULL
`, fixture.workspaceID).Scan(&count))
	return count
}