package message

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

	"github.com/stretchr/testify/require"

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

type hierarchyLifecycleTask struct {
	taskID     string
	scopeID    string
	taskNumber int64
}

type hierarchyLifecycleFixture struct {
	workspaceID string
	boardID     string
	channelID   string
	memberID    string
	parent      hierarchyLifecycleTask
	children    []hierarchyLifecycleTask
}

func (fixture hierarchyLifecycleFixture) allScopeIDs() []string {
	return []string{
		fixture.parent.scopeID,
		fixture.children[0].scopeID,
		fixture.children[1].scopeID,
	}
}

type hierarchyLifecycleGuardResult struct {
	active bool
	err    error
}

type legacyHierarchyDeleteResult struct {
	tombstonedScopeIDs []string
	err                error
}

type hierarchyLifecycleTopologyObservedDB struct {
	coredb.DB
	topologyRead chan<- struct{}
	once         *sync.Once
}

type hierarchyLifecycleTopologyObservedTx struct {
	coredb.Tx
	topologyRead chan<- struct{}
	once         *sync.Once
}

type hierarchyLifecycleTopologyObservedRows struct {
	coredb.Rows
	topologyRead chan<- struct{}
	once         *sync.Once
}

func newHierarchyLifecycleTopologyObservedDB(
	db coredb.DB,
	topologyRead chan<- struct{},
) coredb.DB {
	return &hierarchyLifecycleTopologyObservedDB{
		DB:           db,
		topologyRead: topologyRead,
		once:         &sync.Once{},
	}
}

func (db *hierarchyLifecycleTopologyObservedDB) BeginTx(
	ctx context.Context,
	options coredb.TxOptions,
) (coredb.Tx, error) {
	tx, err := db.DB.BeginTx(ctx, options)
	if err != nil {
		return nil, err
	}
	return &hierarchyLifecycleTopologyObservedTx{
		Tx:           tx,
		topologyRead: db.topologyRead,
		once:         db.once,
	}, nil
}

func (tx *hierarchyLifecycleTopologyObservedTx) Query(
	ctx context.Context,
	query string,
	args ...any,
) (coredb.Rows, error) {
	rows, err := tx.Tx.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	if !isTaskScopeLifecycleTopologyHintQuery(query) {
		return rows, nil
	}
	return &hierarchyLifecycleTopologyObservedRows{
		Rows:         rows,
		topologyRead: tx.topologyRead,
		once:         tx.once,
	}, nil
}

func (rows *hierarchyLifecycleTopologyObservedRows) Close() {
	rows.Rows.Close()
	rows.once.Do(func() {
		close(rows.topologyRead)
	})
}

func isTaskScopeLifecycleTopologyHintQuery(query string) bool {
	return strings.Contains(query, "SELECT t.id,") &&
		strings.Contains(query, "t.task_scope_id") &&
		strings.Contains(query, "COALESCE(t.parent_task_ref") &&
		strings.Contains(query, "t.task_scope_id IN")
}

// runLegacyHierarchyDelete mirrors the lifecycle-relevant SQL order in
// taskboard.TxStore.DeleteForUser. releaseAfterTaskLock is a test-only barrier
// used to make the delete-first interleavings deterministic.
func runLegacyHierarchyDelete(
	ctx context.Context,
	db coredb.DB,
	selectedTaskID string,
	attempted chan<- struct{},
	taskLocked chan<- struct{},
	releaseAfterTaskLock <-chan struct{},
) ([]string, error) {
	tombstonedScopeIDs := make([]string, 0, 3)
	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		close(attempted)

		var selectedScopeID string
		if err := tx.QueryRow(ctx, `
SELECT task_scope_id
FROM tasks
WHERE id = $1
FOR UPDATE
`, selectedTaskID).Scan(&selectedScopeID); err != nil {
			return fmt.Errorf("lock selected Task: %w", err)
		}
		close(taskLocked)

		if releaseAfterTaskLock != nil {
			select {
			case <-releaseAfterTaskLock:
			case <-ctx.Done():
				return ctx.Err()
			}
		}

		rows, err := tx.Query(ctx, `
SELECT id, task_scope_id
FROM tasks
WHERE parent_task_ref = $1
ORDER BY task_number ASC, created_at ASC, id ASC
`, selectedTaskID)
		if err != nil {
			return fmt.Errorf("list legacy child Tasks: %w", err)
		}
		type childRow struct {
			taskID  string
			scopeID string
		}
		children := make([]childRow, 0, 2)
		for rows.Next() {
			var child childRow
			if err := rows.Scan(&child.taskID, &child.scopeID); err != nil {
				rows.Close()
				return fmt.Errorf("scan legacy child Task: %w", err)
			}
			children = append(children, child)
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return fmt.Errorf("iterate legacy child Tasks: %w", err)
		}
		rows.Close()

		now := time.Now().UTC()
		for _, child := range children {
			if err := tombstoneHierarchyTaskScope(ctx, tx, child.scopeID, now); err != nil {
				return err
			}
			tombstonedScopeIDs = append(tombstonedScopeIDs, child.scopeID)
		}
		if len(children) > 0 {
			if _, err := tx.Exec(ctx, `
DELETE FROM tasks
WHERE parent_task_ref = $1
`, selectedTaskID); err != nil {
				return fmt.Errorf("delete legacy child Tasks: %w", err)
			}
		}

		if err := tombstoneHierarchyTaskScope(ctx, tx, selectedScopeID, now); err != nil {
			return err
		}
		tombstonedScopeIDs = append(tombstonedScopeIDs, selectedScopeID)
		commandTag, err := tx.Exec(ctx, `
DELETE FROM tasks
WHERE id = $1
`, selectedTaskID)
		if err != nil {
			return fmt.Errorf("delete selected legacy Task: %w", err)
		}
		if commandTag.RowsAffected() != 1 {
			return fmt.Errorf("delete selected legacy Task: got %d affected rows", commandTag.RowsAffected())
		}
		return nil
	})
	return tombstonedScopeIDs, err
}

// runLegacyHierarchyPromote mirrors DeleteForUser's promote mode: it locks R,
// makes every child top-level, tombstones R's scope, and deletes R. In
// particular, it never locks a child before R, which is the rolling-deploy
// interleaving the lifecycle guard must retry rather than misclassify as a
// deleted child.
func runLegacyHierarchyPromote(
	ctx context.Context,
	db coredb.DB,
	parentTaskID string,
	attempted chan<- struct{},
	parentLocked chan<- struct{},
	releaseAfterParentLock <-chan struct{},
) error {
	return coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		close(attempted)

		var parentScopeID string
		if err := tx.QueryRow(ctx, `
SELECT task_scope_id
FROM tasks
WHERE id = $1
FOR UPDATE
`, parentTaskID).Scan(&parentScopeID); err != nil {
			return fmt.Errorf("lock promoted parent Task: %w", err)
		}
		close(parentLocked)

		if releaseAfterParentLock != nil {
			select {
			case <-releaseAfterParentLock:
			case <-ctx.Done():
				return ctx.Err()
			}
		}

		now := time.Now().UTC()
		commandTag, err := tx.Exec(ctx, `
UPDATE tasks
SET parent_task_ref = '',
    updated_at = $2
WHERE parent_task_ref = $1
`, parentTaskID, now)
		if err != nil {
			return fmt.Errorf("promote legacy child Tasks: %w", err)
		}
		if commandTag.RowsAffected() != 2 {
			return fmt.Errorf("promote legacy child Tasks: got %d affected rows", commandTag.RowsAffected())
		}

		if err := tombstoneHierarchyTaskScope(ctx, tx, parentScopeID, now); err != nil {
			return err
		}
		commandTag, err = tx.Exec(ctx, `
DELETE FROM tasks
WHERE id = $1
`, parentTaskID)
		if err != nil {
			return fmt.Errorf("delete promoted parent Task: %w", err)
		}
		if commandTag.RowsAffected() != 1 {
			return fmt.Errorf("delete promoted parent Task: got %d affected rows", commandTag.RowsAffected())
		}
		return nil
	})
}

func tombstoneHierarchyTaskScope(ctx context.Context, tx coredb.Tx, scopeID string, now time.Time) error {
	commandTag, err := tx.Exec(ctx, `
UPDATE scopes
SET status = 'deleted',
    archived_at = COALESCE(archived_at, $2),
    updated_at = $2
WHERE id = $1
  AND type = 'task'
`, scopeID, now)
	if err != nil {
		return fmt.Errorf("tombstone Task scope %q: %w", scopeID, err)
	}
	if commandTag.RowsAffected() != 1 {
		return fmt.Errorf("tombstone Task scope %q: got %d affected rows", scopeID, commandTag.RowsAffected())
	}
	return nil
}

func seedTaskScopeLifecycleHierarchy(
	t *testing.T,
	ctx context.Context,
	db coredb.QueryExecer,
	suffix string,
) hierarchyLifecycleFixture {
	t.Helper()

	now := time.Now().UTC().Add(-time.Minute).Truncate(time.Second)
	fixture := hierarchyLifecycleFixture{
		workspaceID: "wsp_lifecycle_hierarchy_" + suffix,
		boardID:     "tbd_lifecycle_hierarchy_" + suffix,
		channelID:   "chn_lifecycle_hierarchy_" + suffix,
		memberID:    "mem_lifecycle_hierarchy_" + suffix,
		parent: hierarchyLifecycleTask{
			taskID:     "task_lifecycle_parent_" + suffix,
			scopeID:    "scope_lifecycle_parent_" + suffix,
			taskNumber: 1,
		},
		// Deliberately make lexical ID order the reverse of task-number order.
		// The guard uses authoritative Task-ID lock order while the legacy delete
		// tombstones children in task-number order.
		children: []hierarchyLifecycleTask{
			{
				taskID:     "task_lifecycle_z_child_" + suffix,
				scopeID:    "scope_lifecycle_z_child_" + suffix,
				taskNumber: 2,
			},
			{
				taskID:     "task_lifecycle_a_child_" + suffix,
				scopeID:    "scope_lifecycle_a_child_" + suffix,
				taskNumber: 3,
			},
		},
	}
	userID := "usr_lifecycle_hierarchy_" + suffix
	insertMessagingWorkspace(t, ctx, db, fixture.workspaceID, now)
	insertMessagingUser(t, ctx, db, userID, suffix+"@lifecycle.example.com", now)
	insertMessagingHumanMember(t, ctx, db, fixture.memberID, fixture.workspaceID, userID, now)

	require.NoError(t, NewScopeRepositories(db).Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                fixture.channelID,
		WorkspaceID:       fixture.workspaceID,
		Type:              "channel",
		IdentityKey:       "scope/channel/" + fixture.channelID,
		Status:            "active",
		CreatedByMemberID: fixture.memberID,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	mustExecMessaging(t, ctx, db, `
INSERT INTO task_boards (
    channel_id, workspace_id, next_task_number, created_at, updated_at,
    id
) VALUES ($1, $2, 4, $3, $3, $4)
`, fixture.channelID, fixture.workspaceID, now, fixture.boardID)
	// Task conversation materialization reads content authority from the
	// canonical parent Channel membership, not from Task participation rows.
	// Seed that parent relation once for every hierarchy fixture so lifecycle
	// and materialization tests exercise the same production authority boundary.
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $3, $3, NULL, NULL)
`, fixture.channelID, fixture.memberID, now)

	allTasks := append([]hierarchyLifecycleTask{fixture.parent}, fixture.children...)
	for _, task := range allTasks {
		require.NoError(t, NewScopeRepositories(db).Scopes.UpsertScope(ctx, ScopeRecord{
			ID:                task.scopeID,
			WorkspaceID:       fixture.workspaceID,
			Type:              "task",
			IdentityKey:       "scope/task/" + task.scopeID,
			Status:            "active",
			CreatedByMemberID: fixture.memberID,
			CreatedAt:         now.Add(time.Duration(task.taskNumber) * time.Second),
			UpdatedAt:         now.Add(time.Duration(task.taskNumber) * time.Second),
		}))
		parentTaskRef := ""
		if task.taskID != fixture.parent.taskID {
			parentTaskRef = fixture.parent.taskID
		}
		mustExecMessaging(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, parent_task_ref, created_at, updated_at
) VALUES (
    $1, $2, $3, $4, $5, $1, 'todo',
    $6, $6, $7, $8, $8
)
`, task.taskID, task.scopeID, fixture.boardID, fixture.channelID, task.taskNumber, fixture.memberID, parentTaskRef, now.Add(time.Duration(task.taskNumber)*time.Second))
	}
	return fixture
}

func requireHierarchyDeleted(
	t *testing.T,
	ctx context.Context,
	db coredb.QueryRower,
	fixture hierarchyLifecycleFixture,
) {
	t.Helper()
	requireTaskRowCount(t, ctx, db, fixture.parent.taskID, 0)
	requireTaskScopeStatus(t, ctx, db, fixture.parent.scopeID, "deleted")
	for _, child := range fixture.children {
		requireTaskRowCount(t, ctx, db, child.taskID, 0)
		requireTaskScopeStatus(t, ctx, db, child.scopeID, "deleted")
	}
}

func requireDirectChildDeleteState(
	t *testing.T,
	ctx context.Context,
	db coredb.QueryRower,
	fixture hierarchyLifecycleFixture,
	deletedChild hierarchyLifecycleTask,
) {
	t.Helper()
	requireTaskRowCount(t, ctx, db, deletedChild.taskID, 0)
	requireTaskScopeStatus(t, ctx, db, deletedChild.scopeID, "deleted")
	requireTaskRowCount(t, ctx, db, fixture.parent.taskID, 1)
	requireTaskScopeStatus(t, ctx, db, fixture.parent.scopeID, "active")
	for _, child := range fixture.children {
		if child.taskID == deletedChild.taskID {
			continue
		}
		requireTaskRowCount(t, ctx, db, child.taskID, 1)
		requireTaskScopeStatus(t, ctx, db, child.scopeID, "active")
	}
}

func requirePromotedHierarchyState(
	t *testing.T,
	ctx context.Context,
	db coredb.QueryRower,
	fixture hierarchyLifecycleFixture,
) {
	t.Helper()
	requireTaskRowCount(t, ctx, db, fixture.parent.taskID, 0)
	requireTaskScopeStatus(t, ctx, db, fixture.parent.scopeID, "deleted")
	for _, child := range fixture.children {
		requireTaskRowCount(t, ctx, db, child.taskID, 1)
		requireTaskScopeStatus(t, ctx, db, child.scopeID, "active")
		var parentTaskRef string
		require.NoError(t, db.QueryRow(ctx, `
SELECT parent_task_ref
FROM tasks
WHERE id = $1
`, child.taskID).Scan(&parentTaskRef))
		require.Empty(t, parentTaskRef)
	}
}

func requireTaskRowCount(t *testing.T, ctx context.Context, db coredb.QueryRower, taskID string, want int64) {
	t.Helper()
	var count int64
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM tasks WHERE id = $1`, taskID).Scan(&count))
	require.Equal(t, want, count)
}

func requireTaskScopeStatus(t *testing.T, ctx context.Context, db coredb.QueryRower, scopeID string, want string) {
	t.Helper()
	var status string
	require.NoError(t, db.QueryRow(ctx, `SELECT status FROM scopes WHERE id = $1`, scopeID).Scan(&status))
	require.Equal(t, want, status)
}

func requireHierarchySignal(t *testing.T, ctx context.Context, signal <-chan struct{}, failure string) {
	t.Helper()
	select {
	case <-signal:
	case <-ctx.Done():
		t.Fatalf("%s: %v", failure, ctx.Err())
	}
}

func requireHierarchyBlocked(
	t *testing.T,
	taskLocked <-chan struct{},
	deleteDone <-chan legacyHierarchyDeleteResult,
	failure string,
) {
	t.Helper()
	select {
	case <-taskLocked:
		t.Fatal(failure)
	case result := <-deleteDone:
		t.Fatalf("%s; delete returned early: %v", failure, result.err)
	case <-time.After(250 * time.Millisecond):
	}
}

func requireHierarchyGuardBlocked(
	t *testing.T,
	callbackStarted <-chan struct{},
	guardDone <-chan hierarchyLifecycleGuardResult,
	failure string,
) {
	t.Helper()
	select {
	case <-callbackStarted:
		t.Fatal(failure)
	case result := <-guardDone:
		t.Fatalf("%s; guard returned early: active=%v err=%v", failure, result.active, result.err)
	case <-time.After(250 * time.Millisecond):
	}
}

func requireHierarchyErrorResult(
	t *testing.T,
	ctx context.Context,
	results <-chan error,
	operation string,
) error {
	t.Helper()
	select {
	case err := <-results:
		return err
	case <-ctx.Done():
		t.Fatalf("waiting for %s: %v", operation, ctx.Err())
		return ctx.Err()
	}
}

func requireHierarchyGuardResult(
	t *testing.T,
	ctx context.Context,
	results <-chan hierarchyLifecycleGuardResult,
) hierarchyLifecycleGuardResult {
	t.Helper()
	select {
	case result := <-results:
		return result
	case <-ctx.Done():
		t.Fatalf("waiting for hierarchy lifecycle guard: %v", ctx.Err())
		return hierarchyLifecycleGuardResult{}
	}
}

func requireLegacyHierarchyDeleteResult(
	t *testing.T,
	ctx context.Context,
	results <-chan legacyHierarchyDeleteResult,
) legacyHierarchyDeleteResult {
	t.Helper()
	select {
	case result := <-results:
		return result
	case <-ctx.Done():
		t.Fatalf("waiting for legacy hierarchy delete: %v", ctx.Err())
		return legacyHierarchyDeleteResult{}
	}
}