package controlplane

import (
	"context"
	"errors"
	"fmt"
	"log"
	"path/filepath"
	"sort"
	"strings"

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// StoppedSourceSetReleaseRepoSource binds one release_repo URI to the exact
// operation-local checkout that contains it.
type StoppedSourceSetReleaseRepoSource struct {
	URI      string `json:"uri"`
	PackRoot string `json:"pack_root"`
	Revision string `json:"revision"`
}

// StoppedSourceSetRequest is the pack set to execute while Runtime is stopped.
// Packs not named by the request remain untouched. OperationID is correlation
// only: this disposable-Test mutation is one-shot and a failed call is not
// resumed, deduplicated, or rolled back. A failure after pack mutation is
// recorded as failed_dirty; the caller must keep Runtime stopped and retry the
// full source set. PackActivation remains the last successfully published
// generation identity, not a rollback snapshot of mutable metadata tables.
type StoppedSourceSetRequest struct {
	OperationID                 string
	PackSources                 []string
	ReleaseRepoSources          []StoppedSourceSetReleaseRepoSource
	OfficialRepoSourceRevisions map[string]string
}

// StoppedSourceSetResult reports only that the requested mutation command
// returned successfully. It is not a Candidate or verification receipt.
type StoppedSourceSetResult struct {
	OperationID string `json:"operation_id"`
	Applied     bool   `json:"applied"`
}

func ApplyStoppedSourceSetWithRunnerOptions(
	ctx context.Context,
	db *database.TiDBConnection,
	request StoppedSourceSetRequest,
	runnerOptions RunnerOptions,
) (StoppedSourceSetResult, error) {
	operationID := strings.TrimSpace(request.OperationID)
	if operationID == "" {
		return StoppedSourceSetResult{}, fmt.Errorf("stopped source-set operation ID is required")
	}
	if db == nil || db.DB() == nil {
		return StoppedSourceSetResult{}, fmt.Errorf("stopped source-set database connection is unavailable")
	}
	set, err := resolveStoppedSourceSet(request, runnerOptions.ComponentKindRegistrationSet)
	if err != nil {
		return StoppedSourceSetResult{}, err
	}
	runtimeOverrides, err := stoppedSourceSetRuntimeOverrides(set)
	if err != nil {
		return StoppedSourceSetResult{}, fmt.Errorf("construct stopped source-set runtime context: %w", err)
	}
	currentDB := db
	err = withControlPlaneApplyLockScope(
		ctx,
		db.DB(),
		"ApplyStoppedSourceSet",
		func(scope controlPlaneApplyLockScope) (operationErr error) {
			if err := requireControlPlaneApplyLockScope(scope); err != nil {
				return err
			}
			applyCtx := packs.WithOperationPackSourceOptions(
				scope.identity.leaseContext,
				set.sourceOptions,
			)
			targetPacks := make([]packs.Pack, 0, len(set.targets))
			for _, target := range set.targets {
				targetPacks = append(targetPacks, target.pack)
			}
			admissions, err := prepareStoppedSourceSetReplacementGenerations(
				applyCtx,
				currentDB.DB(),
				targetPacks,
				scope,
			)
			if err != nil {
				return fmt.Errorf("admit stopped source-set replacement lifecycle: %w", err)
			}
			// Candidate preparation commits independently, so every later exit must
			// close those attempts. Install this guard before any other fallible
			// discovery; otherwise an early read error can strand `preparing` forever.
			markerInvalidated := false
			defer func() {
				markerDB := db.DB()
				if currentDB != nil && currentDB.DB() != nil {
					markerDB = currentDB.DB()
				}
				if operationErr != nil && markerInvalidated {
					if markErr := bootstrapStateStore.MarkControlPlaneApplyAllFailed(markerDB, operationErr); markErr != nil {
						operationErr = errors.Join(operationErr, fmt.Errorf("mark ApplyAll failed after stopped source-set error: %w", markErr))
					}
				}
				if operationErr != nil {
					for _, admission := range admissions {
						if markErr := markStoppedSourceSetPackGenerationFailed(
							context.WithoutCancel(applyCtx),
							markerDB,
							admission.packName,
							admission.candidatePackInstallID,
							false,
							operationErr,
						); markErr != nil {
							operationErr = errors.Join(operationErr, markErr)
						}
					}
				}
				if currentDB != nil && currentDB != db {
					if closeErr := currentDB.Close(); closeErr != nil {
						operationErr = errors.Join(operationErr, fmt.Errorf("close stopped source-set refreshed database connection: %w", closeErr))
					}
				}
			}()
			excludedAttributionPeers := make(map[string]struct{}, len(set.targets))
			for _, target := range set.targets {
				excludedAttributionPeers[strings.ToLower(strings.TrimSpace(target.pack.Name()))] = struct{}{}
			}
			runtimeOverrides.activeGenerationAttributionPeers, err = loadActiveGenerationAttributionPeerContext(
				applyCtx,
				currentDB.DB(),
				excludedAttributionPeers,
			)
			if err != nil {
				return fmt.Errorf("load stopped source-set attribution authority: %w", err)
			}
			if bootstrapStateStore == nil {
				return fmt.Errorf("stopped source-set ApplyAll marker store is unavailable")
			}
			if err := bootstrapStateStore.MarkControlPlaneApplyAllInvalidated(
				currentDB.DB(), "ApplyStoppedSourceSet",
			); err != nil {
				return fmt.Errorf("invalidate ApplyAll completion marker before stopped source-set mutation: %w", err)
			}
			markerInvalidated = true
			log.Printf("controlplane: invalidated full ApplyAll marker before stopped source-set mutation; only full ApplyAll may restore completed")
			currentDB, err = ApplyMetadata(applyCtx, currentDB)
			if err != nil {
				return fmt.Errorf("apply stopped source-set substrate metadata: %w", err)
			}
			runner, err := NewRunnerWithOptions(currentDB, runnerOptions)
			if err != nil {
				return fmt.Errorf("construct stopped source-set runner: %w", err)
			}
			if err := runner.RefreshCache(applyCtx); err != nil {
				return fmt.Errorf("refresh stopped source-set runner: %w", err)
			}
			authority := offlinecandidateauthority.NewStoppedSourceSetAuthority()
			for _, target := range set.targets {
				admission := admissions[strings.ToLower(strings.TrimSpace(target.pack.Name()))]
				if err := applyResolvedSourceBackedPackUnlockedWithPreflightOverrides(
					applyCtx,
					runner,
					target.pack,
					runtimeOverrides,
					InstallPackOptions{
						stoppedSourceSetApplyOnly:              true,
						stoppedSourceSetAuthority:              authority,
						stoppedSourceSetPreviousPackInstallID:  admission.activePackInstallID,
						stoppedSourceSetCandidatePackInstallID: admission.candidatePackInstallID,
					},
				); err != nil {
					return fmt.Errorf("apply stopped source-set pack %s: %w", target.pack.Name(), err)
				}
			}
			return nil
		},
	)
	if err != nil {
		return StoppedSourceSetResult{}, err
	}
	return StoppedSourceSetResult{OperationID: operationID, Applied: true}, nil
}

type stoppedSourceSetTarget struct {
	pack *packs.SourceBackedPack
}

type resolvedStoppedSourceSet struct {
	targets       []stoppedSourceSetTarget
	byName        map[string]stoppedSourceSetTarget
	sourceOptions packs.PackSourceOptions
}

func resolveStoppedSourceSet(
	request StoppedSourceSetRequest,
	registrations packs.ComponentKindRegistrationSet,
) (resolvedStoppedSourceSet, error) {
	if len(request.PackSources) == 0 {
		return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set requires at least one pack source")
	}
	sourceOptions := packs.DefaultPackSourceOptions()
	sourceOptions.OfficialRepoSourceRevisions = make(map[string]string, len(request.OfficialRepoSourceRevisions))
	for rawName, rawRevision := range request.OfficialRepoSourceRevisions {
		name := strings.ToLower(strings.TrimSpace(rawName))
		revision := strings.TrimSpace(rawRevision)
		if name == "" || revision == "" {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set official source revision is incomplete")
		}
		if err := packs.ValidateReleaseRepoRevision(revision); err != nil {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set official source %s: %w", name, err)
		}
		if _, duplicate := sourceOptions.OfficialRepoSourceRevisions[name]; duplicate {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set official source %s is duplicated", name)
		}
		sourceOptions.OfficialRepoSourceRevisions[name] = revision
	}
	sourceOptions.ReleaseRepoSources = make(map[string]packs.ReleaseRepoSource, len(request.ReleaseRepoSources))
	for _, authority := range request.ReleaseRepoSources {
		uri := strings.TrimSpace(authority.URI)
		root := strings.TrimSpace(authority.PackRoot)
		revision := strings.TrimSpace(authority.Revision)
		if uri == "" || root == "" || !filepath.IsAbs(root) || filepath.Clean(root) != root {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set release source %q has no canonical absolute root", uri)
		}
		if err := packs.ValidateReleaseRepoRevision(revision); err != nil {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set release source %s: %w", uri, err)
		}
		if _, duplicate := sourceOptions.ReleaseRepoSources[uri]; duplicate {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set release source %s is duplicated", uri)
		}
		sourceOptions.ReleaseRepoSources[uri] = packs.ReleaseRepoSource{Root: root, Revision: revision}
	}

	resolvedSources := make([]*packs.ResolvedPackSource, 0, len(request.PackSources))
	var formulaBundles []formula.FunctionBundle
	for index, rawSource := range request.PackSources {
		ref, err := packs.ParsePackSourceRef(rawSource)
		if err != nil {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set source %d: %w", index, err)
		}
		switch ref.Kind {
		case packs.PackSourceKindOfficialRepo:
			if strings.TrimSpace(sourceOptions.OfficialRepoSourceRevisions[strings.ToLower(ref.URI)]) == "" {
				return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set official source %s has no exact revision", ref.URI)
			}
		case packs.PackSourceKindReleaseRepo:
			if _, ok := sourceOptions.ReleaseRepoSources[ref.URI]; !ok {
				return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set release source %s has no exact authority", ref.URI)
			}
		default:
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set source %d must be official_repo or release_repo", index)
		}
		report := packs.ValidatePackSourceWithComponentKindRegistrationSet(
			rawSource,
			sourceOptions,
			registrations,
		)
		if report.Source == nil || report.Source.Manifest == nil || report.HasErrors() {
			return resolvedStoppedSourceSet{}, fmt.Errorf(
				"stopped source-set source %d is invalid: %s",
				index, report.Error(),
			)
		}
		resolved := report.Source
		if ref.Kind == packs.PackSourceKindOfficialRepo && strings.TrimSpace(resolved.Manifest.Name) != ref.URI {
			return resolvedStoppedSourceSet{}, fmt.Errorf(
				"stopped source-set official source %q resolved manifest name %q",
				ref.URI, resolved.Manifest.Name,
			)
		}
		bundles, err := packs.RuntimeFormulaFunctionBundlesFromResolvedSource(resolved)
		if err != nil {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set source %d formula functions: %w", index, err)
		}
		formulaBundles = append(formulaBundles, bundles...)
		resolvedSources = append(resolvedSources, resolved)
	}

	set := resolvedStoppedSourceSet{
		targets:       make([]stoppedSourceSetTarget, 0, len(resolvedSources)),
		byName:        make(map[string]stoppedSourceSetTarget, len(resolvedSources)),
		sourceOptions: sourceOptions,
	}
	for index, source := range resolvedSources {
		pack, err := packs.NewSourceBackedPack(
			source,
			packs.WithStoppedSourceSetApplyOnly(
				offlinecandidateauthority.NewStoppedSourceSetAuthority(),
			),
			packs.WithComponentKindRegistrationSet(registrations),
			packs.WithRuntimeFormulaFunctionBundlesForValidation(
				packs.RuntimeFormulaFunctionBundlesExcludingResolvedSource(source, formulaBundles),
			),
			packs.WithFlowActionCallRuntimeFromPackSourceOptions(sourceOptions),
		)
		if err != nil {
			return resolvedStoppedSourceSet{}, fmt.Errorf("construct stopped source-set source %d: %w", index, err)
		}
		name := strings.ToLower(strings.TrimSpace(pack.Name()))
		if name == "" {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set source %d has no pack name", index)
		}
		if _, duplicate := set.byName[name]; duplicate {
			return resolvedStoppedSourceSet{}, fmt.Errorf("stopped source-set pack %s is duplicated", pack.Name())
		}
		target := stoppedSourceSetTarget{pack: pack}
		set.targets = append(set.targets, target)
		set.byName[name] = target
	}
	orderedTargets, err := orderStoppedSourceSetTargets(set)
	if err != nil {
		return resolvedStoppedSourceSet{}, err
	}
	set.targets = orderedTargets
	return set, nil
}

// orderStoppedSourceSetTargets orders only dependencies that are replaced by
// this request. Dependencies outside the set are left to the real pack apply
// path instead of being admitted by a stopped-deployment preflight.
func orderStoppedSourceSetTargets(set resolvedStoppedSourceSet) ([]stoppedSourceSetTarget, error) {
	if len(set.targets) != len(set.byName) {
		return nil, fmt.Errorf("stopped source-set target set is internally inconsistent")
	}
	inputOrder := make(map[string]int, len(set.targets))
	indegree := make(map[string]int, len(set.targets))
	dependents := make(map[string][]string, len(set.targets))
	for index, target := range set.targets {
		if target.pack == nil {
			return nil, fmt.Errorf("stopped source-set target %d is empty", index)
		}
		name := strings.ToLower(strings.TrimSpace(target.pack.Name()))
		if name == "" {
			return nil, fmt.Errorf("stopped source-set target %d has no pack name", index)
		}
		mapped, ok := set.byName[name]
		if !ok || mapped.pack != target.pack {
			return nil, fmt.Errorf("stopped source-set target %s does not match its name index", target.pack.Name())
		}
		inputOrder[name] = index
		indegree[name] = 0
	}
	for _, target := range set.targets {
		name := strings.ToLower(strings.TrimSpace(target.pack.Name()))
		for _, dependency := range target.pack.PackDependencies() {
			dependencyName := strings.ToLower(strings.TrimSpace(dependency.Name))
			if dependencyName == "" {
				return nil, fmt.Errorf("stopped source-set pack %s has an empty dependency", target.pack.Name())
			}
			if _, replaced := set.byName[dependencyName]; !replaced {
				continue
			}
			indegree[name]++
			dependents[dependencyName] = append(dependents[dependencyName], name)
		}
	}

	ready := make([]string, 0, len(set.targets))
	for _, target := range set.targets {
		name := strings.ToLower(strings.TrimSpace(target.pack.Name()))
		if indegree[name] == 0 {
			ready = append(ready, name)
		}
	}
	ordered := make([]stoppedSourceSetTarget, 0, len(set.targets))
	for len(ready) > 0 {
		name := ready[0]
		ready = ready[1:]
		ordered = append(ordered, set.byName[name])
		for _, dependent := range dependents[name] {
			indegree[dependent]--
			if indegree[dependent] == 0 {
				ready = append(ready, dependent)
			}
		}
		sort.SliceStable(ready, func(i, j int) bool {
			return inputOrder[ready[i]] < inputOrder[ready[j]]
		})
	}
	if len(ordered) != len(set.targets) {
		return nil, fmt.Errorf("stopped source-set pack dependency graph contains a cycle")
	}
	return ordered, nil
}

func stoppedSourceSetRuntimeOverrides(set resolvedStoppedSourceSet) (*installPackPreflightOverrides, error) {
	targets := make(map[string]*packs.SourceBackedPack, len(set.targets))
	for _, target := range set.targets {
		targets[strings.ToLower(strings.TrimSpace(target.pack.Name()))] = target.pack
	}
	desired, err := newInstallPackPreflightPackContext(targets)
	if err != nil {
		return nil, err
	}
	runtime, err := newInstallPackPreflightPackContext(targets)
	if err != nil {
		return nil, err
	}
	return &installPackPreflightOverrides{
		plannedDesiredPacks:   desired,
		runtimeEffectivePacks: runtime,
	}, nil
}