package message

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

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

	dsn := testkit.StartInitializedTiDBWithRootDSN(t, messagingTiDB.RootDSN(t))
	ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
	defer cancel()

	sqlDB, err := testkit.OpenTiDBContext(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(func() { _ = sqlDB.Close() })
	sqlDB.SetMaxOpenConns(4)
	sqlDB.SetMaxIdleConns(4)
	db := coredb.WrapSQLDB(sqlDB)

	const (
		workspaceID = "wsp_task_scope_lifecycle_lock"
		scopeID     = "tsk_task_scope_lifecycle_lock"
	)
	seedActiveTaskScopeForLifecycleTest(t, ctx, db, workspaceID, scopeID)
	store := New(db, nil, 0, Options{})

	effectStarted := make(chan struct{})
	releaseEffect := make(chan struct{})
	guardDone := make(chan taskScopeLifecycleResult, 1)
	var effects atomic.Int64
	go func() {
		active, guardErr := store.WithActiveTaskScopes(ctx, workspaceID, []string{scopeID}, func() error {
			close(effectStarted)
			select {
			case <-releaseEffect:
			case <-ctx.Done():
				return ctx.Err()
			}
			effects.Add(1)
			return nil
		})
		guardDone <- taskScopeLifecycleResult{active: active, err: guardErr}
	}()

	select {
	case <-effectStarted:
	case <-ctx.Done():
		t.Fatalf("Task fanout callback did not start: %v", ctx.Err())
	}

	deleteAttempted := make(chan struct{})
	deleteDone := make(chan error, 1)
	go func() {
		deleteDone <- coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
			close(deleteAttempted)
			var lockedTaskID string
			if lockErr := tx.QueryRow(ctx, `
SELECT id
FROM tasks
WHERE id = $1
FOR UPDATE
`, scopeID).Scan(&lockedTaskID); lockErr != nil {
				return lockErr
			}
			_, updateErr := tx.Exec(ctx, `
UPDATE scopes
SET status = 'deleted',
    archived_at = $3,
    updated_at = $3
WHERE workspace_id = $1
  AND id = $2
`, workspaceID, scopeID, time.Now().UTC())
			if updateErr != nil {
				return updateErr
			}
			_, deleteErr := tx.Exec(ctx, `DELETE FROM tasks WHERE id = $1`, scopeID)
			return deleteErr
		})
	}()

	select {
	case <-deleteAttempted:
	case <-ctx.Done():
		t.Fatalf("Task delete did not reach the lifecycle update: %v", ctx.Err())
	}

	// The delete transaction updates the same scope row and must remain blocked
	// until the callback completes all unread/stream/wake side effects.
	select {
	case deleteErr := <-deleteDone:
		t.Fatalf("Task delete completed while fanout effects were still running: %v", deleteErr)
	case <-time.After(250 * time.Millisecond):
	}

	close(releaseEffect)
	result := <-guardDone
	require.NoError(t, result.err)
	require.True(t, result.active)
	require.Equal(t, int64(1), effects.Load())
	require.NoError(t, <-deleteDone)

	var status string
	var archivedAt *time.Time
	require.NoError(t, db.QueryRow(ctx, `
SELECT status, archived_at
FROM scopes
WHERE workspace_id = $1
  AND id = $2
`, workspaceID, scopeID).Scan(&status, &archivedAt))
	require.Equal(t, "deleted", status)
	require.NotNil(t, archivedAt)
}

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

	ctx := t.Context()
	db := startMessagingTestPool(t, ctx)
	const (
		workspaceID = "wsp_task_scope_lifecycle_deleted"
		scopeID     = "tsk_task_scope_lifecycle_deleted"
		activeID    = "tsk_task_scope_lifecycle_active"
	)
	seedActiveTaskScopeForLifecycleTest(t, ctx, db, workspaceID, scopeID)
	now := time.Now().UTC()
	require.NoError(t, NewScopeRepositories(db).Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                activeID,
		WorkspaceID:       workspaceID,
		Type:              "task",
		IdentityKey:       "task:" + activeID,
		Status:            "active",
		CreatedByMemberID: "mem_" + scopeID,
		CreatedAt:         now.Add(-time.Minute),
		UpdatedAt:         now.Add(-time.Minute),
	}))
	_, err := db.Exec(ctx, `
INSERT INTO tasks (
    id, task_scope_id, board_id, channel_id, task_number, title, status,
    created_by_member_id, created_by_ref, execution_owner_member_id,
    parent_task_ref, created_at, updated_at
) VALUES (
    $1, $1, $2, $3, 2, 'Active lifecycle Task', 'todo',
    $4, $4, $4, '', $5, $5
)
`, activeID, "tbd_"+scopeID, "chn_"+scopeID, "mem_"+scopeID, now)
	require.NoError(t, err)
	_, err = db.Exec(ctx, `
UPDATE scopes
SET status = 'deleted',
    archived_at = $3,
    updated_at = $3
WHERE workspace_id = $1
  AND id = $2
`, workspaceID, scopeID, now)
	require.NoError(t, err)

	var callbacks atomic.Int64
	active, err := New(db, nil, 0, Options{}).WithActiveTaskScopes(ctx, workspaceID, []string{scopeID, activeID}, func() error {
		callbacks.Add(1)
		return nil
	})
	require.NoError(t, err)
	require.False(t, active)
	require.Zero(t, callbacks.Load())
}

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

	ctx := t.Context()
	db := startMessagingTestPool(t, ctx)
	const (
		workspaceID = "wsp_task_scope_lifecycle_legacy_board"
		scopeID     = "tsk_task_scope_lifecycle_legacy_board"
	)
	seedActiveTaskScopeForLifecycleTest(t, ctx, db, workspaceID, scopeID)
	_, err := db.Exec(ctx, `
UPDATE tasks
SET board_id = NULL
WHERE id = $1
`, scopeID)
	require.NoError(t, err)

	var callbacks atomic.Int64
	active, err := New(db, nil, 0, Options{}).WithActiveTaskScopes(ctx, workspaceID, []string{scopeID}, func() error {
		callbacks.Add(1)
		return nil
	})
	require.NoError(t, err)
	require.False(t, active)
	require.Zero(t, callbacks.Load())
}

type taskScopeLifecycleResult struct {
	active bool
	err    error
}

func seedActiveTaskScopeForLifecycleTest(
	t *testing.T,
	ctx context.Context,
	db coredb.QueryExecer,
	workspaceID string,
	scopeID string,
) {
	t.Helper()

	now := time.Now().UTC().Add(-time.Minute).Truncate(time.Second)
	userID := "usr_" + scopeID
	memberID := "mem_" + scopeID
	insertMessagingWorkspace(t, ctx, db, workspaceID, now)
	insertMessagingUser(t, ctx, db, userID, scopeID+"@example.com", now)
	insertMessagingHumanMember(t, ctx, db, memberID, workspaceID, userID, now)
	require.NoError(t, NewScopeRepositories(db).Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                scopeID,
		WorkspaceID:       workspaceID,
		Type:              "task",
		IdentityKey:       "task:" + scopeID,
		Status:            "active",
		CreatedByMemberID: memberID,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	channelID := "chn_" + scopeID
	boardID := "tbd_" + scopeID
	require.NoError(t, NewScopeRepositories(db).Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                channelID,
		WorkspaceID:       workspaceID,
		Type:              "channel",
		IdentityKey:       "channel:" + channelID,
		Status:            "active",
		CreatedByMemberID: memberID,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	_, err := db.Exec(ctx, `
INSERT INTO task_boards (
    channel_id, workspace_id, next_task_number, created_at, updated_at,
    id
) VALUES ($1, $2, 2, $3, $3, $4)
`, channelID, workspaceID, now, boardID)
	require.NoError(t, err)
	_, err = db.Exec(ctx, `
INSERT INTO tasks (
    id, task_scope_id, board_id, channel_id, task_number, title, status,
    created_by_member_id, created_by_ref, execution_owner_member_id,
    parent_task_ref, created_at, updated_at
) VALUES (
    $1, $1, $2, $3, 1, 'Lifecycle Task', 'todo',
    $4, $4, $4, '', $5, $5
)
`, scopeID, boardID, channelID, memberID, now)
	require.NoError(t, err)
}