package message

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"

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

const taskScopeLifecycleTopologyRetryLimit = 8

var errTaskScopeLifecycleTopologyChanged = errors.New("task scope lifecycle topology changed")

type taskScopeLifecycleHint struct {
	taskID       string
	taskScopeID  string
	parentTaskID string
}

// WithActiveTaskScopes holds lifecycle row locks on the Task records that own
// the requested scopes while fanout runs. Task delete already acquires those
// rows before it tombstones scopes, so the Task row is the rolling-compatible
// lifecycle serialization point.
//
// Parents are always locked before children. A plain pre-read supplies topology
// hints, then every top-level root is point-locked in deterministic order before
// any requested child. The locked rows are revalidated; a concurrent reparent
// rolls the whole transaction back and retries rather than acquiring a newly
// discovered parent while holding a child. This keeps the lock order compatible
// with legacy parent-first delete/promote transactions.
//
// The callback must not re-lock the protected Task scope row. DeliveryFanout
// marks the exact protected scopes in its callback context so nested IM
// projection writes can lock member-local state while reading scopes without a
// locking read.
func (store *Store) WithActiveTaskScopes(
	ctx context.Context,
	workspaceID string,
	scopeIDs []string,
	fanout func() error,
) (bool, error) {
	if store == nil || store.db == nil {
		return false, fmt.Errorf("message store is not configured")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	scopeIDs = normalizedTaskScopeLifecycleIDs(scopeIDs)
	if workspaceID == "" || len(scopeIDs) == 0 {
		return false, fmt.Errorf("task scope lifecycle workspace and scopes are required")
	}
	if fanout == nil {
		return false, fmt.Errorf("task scope lifecycle fanout callback is required")
	}

	for attempt := 0; attempt < taskScopeLifecycleTopologyRetryLimit; attempt++ {
		active := false
		err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
			hints, available, err := loadTaskScopeLifecycleHints(ctx, tx, workspaceID, scopeIDs)
			if err != nil {
				return err
			}
			if !available {
				return nil
			}

			lockedRoots, err := lockTaskScopeLifecycleRoots(ctx, tx, workspaceID, hints)
			if err != nil {
				return err
			}
			if lockedRoots == nil {
				return nil
			}

			available, err = lockTaskScopeLifecycleChildren(ctx, tx, workspaceID, hints, lockedRoots)
			if err != nil {
				return err
			}
			if !available {
				return nil
			}

			available, err = taskScopeLifecycleScopesAreActive(ctx, tx, workspaceID, scopeIDs)
			if err != nil {
				return err
			}
			if !available {
				return nil
			}

			active = true
			return fanout()
		})
		if errors.Is(err, errTaskScopeLifecycleTopologyChanged) {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return false, ctxErr
			}
			continue
		}
		if err != nil {
			return active, err
		}
		return active, nil
	}
	return false, fmt.Errorf(
		"task scope lifecycle topology did not stabilize after %d attempts: %w",
		taskScopeLifecycleTopologyRetryLimit,
		errTaskScopeLifecycleTopologyChanged,
	)
}

func loadTaskScopeLifecycleHints(
	ctx context.Context,
	queryer coredb.Queryer,
	workspaceID string,
	scopeIDs []string,
) ([]taskScopeLifecycleHint, bool, error) {
	placeholders := make([]string, 0, len(scopeIDs))
	args := make([]any, 0, len(scopeIDs)+1)
	args = append(args, workspaceID)
	for _, scopeID := range scopeIDs {
		args = append(args, scopeID)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
SELECT t.id,
       t.task_scope_id,
       COALESCE(t.parent_task_ref, '')
FROM tasks t
JOIN task_boards tb
  ON tb.id = t.board_id
 AND tb.workspace_id = $1
JOIN scopes task_channel
  ON task_channel.id = tb.channel_id
 AND task_channel.workspace_id = $1
 AND task_channel.type = 'channel'
 AND task_channel.status = 'active'
 AND task_channel.archived_at IS NULL
 AND t.channel_id = tb.channel_id
WHERE tb.workspace_id = $1
  AND t.task_scope_id IN (%s)
ORDER BY t.task_scope_id ASC
`, strings.Join(placeholders, ", ")), args...)
	if err != nil {
		return nil, false, fmt.Errorf("load task scope lifecycle topology: %w", err)
	}
	defer rows.Close()

	hints := make([]taskScopeLifecycleHint, 0, len(scopeIDs))
	for rows.Next() {
		var hint taskScopeLifecycleHint
		if err := rows.Scan(&hint.taskID, &hint.taskScopeID, &hint.parentTaskID); err != nil {
			return nil, false, fmt.Errorf("scan task scope lifecycle topology: %w", err)
		}
		hint.taskID = strings.TrimSpace(hint.taskID)
		hint.taskScopeID = strings.TrimSpace(hint.taskScopeID)
		hint.parentTaskID = strings.TrimSpace(hint.parentTaskID)
		hints = append(hints, hint)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("iterate task scope lifecycle topology: %w", err)
	}
	if len(hints) != len(scopeIDs) {
		return nil, false, nil
	}
	for index := range scopeIDs {
		if hints[index].taskScopeID != scopeIDs[index] {
			return nil, false, nil
		}
	}
	return hints, true, nil
}

// lockTaskScopeLifecycleRoots returns nil only when a requested root
// disappeared (delete won). If an unrequested referenced parent disappeared,
// it returns the topology sentinel so a concurrent promotion can be observed
// by retrying from an empty lock set.
func lockTaskScopeLifecycleRoots(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	hints []taskScopeLifecycleHint,
) (map[string]struct{}, error) {
	rootIDs := make(map[string]struct{}, len(hints))
	targetTaskIDs := make(map[string]struct{}, len(hints))
	for _, hint := range hints {
		targetTaskIDs[hint.taskID] = struct{}{}
		rootID := hint.parentTaskID
		if rootID == "" {
			rootID = hint.taskID
		}
		rootIDs[rootID] = struct{}{}
	}
	orderedRootIDs := sortedStringSet(rootIDs)
	for _, rootID := range orderedRootIDs {
		var lockedTaskID string
		var parentTaskID string
		var boardID string
		var channelID string
		err := queryer.QueryRow(ctx, `
SELECT t.id,
       COALESCE(t.parent_task_ref, ''),
       COALESCE(t.board_id, ''),
       COALESCE(tb.channel_id, '')
FROM tasks t
JOIN task_boards tb
  ON tb.id = t.board_id
WHERE t.id = $1
FOR UPDATE
		`, rootID).Scan(&lockedTaskID, &parentTaskID, &boardID, &channelID)
		if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
			// A referenced parent can disappear because legacy promote first
			// turns the requested child into a top-level Task and then deletes
			// the old parent. Retry so the surviving child becomes its own
			// root. If the missing root was itself requested, the target was
			// deleted and the whole fanout must remain suppressed.
			if _, requested := targetTaskIDs[rootID]; requested {
				return nil, nil
			}
			return nil, errTaskScopeLifecycleTopologyChanged
		}
		if err != nil {
			return nil, fmt.Errorf("lock task scope lifecycle root %q: %w", rootID, err)
		}
		authorized, err := taskScopeLifecycleBoardBelongsToWorkspace(ctx, queryer, workspaceID, boardID, channelID)
		if err != nil {
			return nil, err
		}
		if !authorized {
			if _, requested := targetTaskIDs[rootID]; requested {
				return nil, nil
			}
			return nil, errTaskScopeLifecycleTopologyChanged
		}
		if strings.TrimSpace(lockedTaskID) != rootID || strings.TrimSpace(parentTaskID) != "" {
			return nil, errTaskScopeLifecycleTopologyChanged
		}
	}
	return rootIDs, nil
}

func lockTaskScopeLifecycleChildren(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	hints []taskScopeLifecycleHint,
	lockedRoots map[string]struct{},
) (bool, error) {
	children := make([]taskScopeLifecycleHint, 0, len(hints))
	for _, hint := range hints {
		if hint.parentTaskID != "" {
			children = append(children, hint)
		}
	}
	sort.Slice(children, func(i, j int) bool {
		if children[i].taskID == children[j].taskID {
			return children[i].taskScopeID < children[j].taskScopeID
		}
		return children[i].taskID < children[j].taskID
	})

	for _, hint := range children {
		var taskScopeID string
		var parentTaskID string
		var boardID string
		var channelID string
		err := queryer.QueryRow(ctx, `
SELECT t.task_scope_id,
       COALESCE(t.parent_task_ref, ''),
       COALESCE(t.board_id, ''),
       COALESCE(tb.channel_id, '')
FROM tasks t
JOIN task_boards tb
  ON tb.id = t.board_id
WHERE t.id = $1
FOR UPDATE
`, hint.taskID).Scan(&taskScopeID, &parentTaskID, &boardID, &channelID)
		if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
			return false, nil
		}
		if err != nil {
			return false, fmt.Errorf("lock task scope lifecycle child %q: %w", hint.taskID, err)
		}
		authorized, err := taskScopeLifecycleBoardBelongsToWorkspace(ctx, queryer, workspaceID, boardID, channelID)
		if err != nil {
			return false, err
		}
		if !authorized {
			return false, nil
		}
		taskScopeID = strings.TrimSpace(taskScopeID)
		parentTaskID = strings.TrimSpace(parentTaskID)
		if taskScopeID != hint.taskScopeID {
			return false, errTaskScopeLifecycleTopologyChanged
		}
		if _, ok := lockedRoots[parentTaskID]; !ok || parentTaskID == "" {
			return false, errTaskScopeLifecycleTopologyChanged
		}
	}
	return true, nil
}

func taskScopeLifecycleBoardBelongsToWorkspace(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	boardID string,
	channelID string,
) (bool, error) {
	boardID = strings.TrimSpace(boardID)
	channelID = strings.TrimSpace(channelID)
	if boardID == "" {
		return false, nil
	}
	var boardWorkspaceID string
	var boardChannelID string
	err := queryer.QueryRow(ctx, `
SELECT tb.workspace_id,
       tb.channel_id
FROM task_boards tb
JOIN scopes task_channel
  ON task_channel.id = tb.channel_id
 AND task_channel.workspace_id = tb.workspace_id
 AND task_channel.type = 'channel'
 AND task_channel.status = 'active'
 AND task_channel.archived_at IS NULL
WHERE tb.id = $1
  AND tb.channel_id = $2
`, boardID, channelID).Scan(&boardWorkspaceID, &boardChannelID)
	if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("validate task scope lifecycle board workspace: %w", err)
	}
	return strings.TrimSpace(boardWorkspaceID) == workspaceID && strings.TrimSpace(boardChannelID) == channelID, nil
}

func taskScopeLifecycleScopesAreActive(
	ctx context.Context,
	queryer coredb.Queryer,
	workspaceID string,
	scopeIDs []string,
) (bool, error) {
	placeholders := make([]string, 0, len(scopeIDs))
	args := make([]any, 0, len(scopeIDs)+1)
	args = append(args, workspaceID)
	for _, scopeID := range scopeIDs {
		args = append(args, scopeID)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
SELECT id
FROM scopes
WHERE workspace_id = $1
  AND id IN (%s)
  AND type = 'task'
  AND status = 'active'
  AND archived_at IS NULL
ORDER BY id ASC
`, strings.Join(placeholders, ", ")), args...)
	if err != nil {
		return false, fmt.Errorf("read active task scope lifecycle: %w", err)
	}
	defer rows.Close()

	activeScopeIDs := make([]string, 0, len(scopeIDs))
	for rows.Next() {
		var scopeID string
		if err := rows.Scan(&scopeID); err != nil {
			return false, fmt.Errorf("scan active task scope lifecycle: %w", err)
		}
		activeScopeIDs = append(activeScopeIDs, strings.TrimSpace(scopeID))
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("iterate active task scope lifecycle: %w", err)
	}
	if len(activeScopeIDs) != len(scopeIDs) {
		return false, nil
	}
	for index := range scopeIDs {
		if activeScopeIDs[index] != scopeIDs[index] {
			return false, nil
		}
	}
	return true, nil
}

func normalizedTaskScopeLifecycleIDs(scopeIDs []string) []string {
	unique := make(map[string]struct{}, len(scopeIDs))
	for _, scopeID := range scopeIDs {
		if scopeID = strings.TrimSpace(scopeID); scopeID != "" {
			unique[scopeID] = struct{}{}
		}
	}
	return sortedStringSet(unique)
}

func sortedStringSet(values map[string]struct{}) []string {
	ordered := make([]string, 0, len(values))
	for value := range values {
		ordered = append(ordered, value)
	}
	sort.Strings(ordered)
	return ordered
}