package controlplane

import (
	"context"
	"fmt"
	"strings"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// resolveApplyAllSourceRevisionAdvanceReplayTuple reconstructs the immutable
// source side of every transition. For an ordinary advance it is the current
// InstalledPack revision. For a retry it is the plan's FromRevision, because
// Begin has already moved the mutable InstalledPack row to ToRevision while the
// active PackGeneration deliberately remains at FromRevision until activation.
func resolveApplyAllSourceRevisionAdvanceReplayTuple(
	ctx context.Context,
	r *Runner,
	snapshot applyAllSourceRevisionAdvanceCurrentSnapshot,
	activeRuntime applyAllSourceRevisionAdvanceActiveRuntimeObservation,
	current applyAllSourceRevisionAdvanceTuple,
) (applyAllSourceRevisionAdvanceTuple, error) {
	if len(snapshot.entries) == 0 {
		return applyAllSourceRevisionAdvanceTuple{
			plan: []normalizedApplyAllSourceRevisionAdvance{}, targets: map[string]*packs.SourceBackedPack{},
		}, nil
	}
	hasRetry := false
	for _, entry := range snapshot.entries {
		if entry.pending != nil && entry.pending.Kind == ApplyAllSourceRevisionAdvancePendingRetry {
			hasRetry = true
			break
		}
	}
	if !hasRetry {
		return cloneApplyAllSourceRevisionAdvanceTuple(current), nil
	}

	plan := make([]normalizedApplyAllSourceRevisionAdvance, 0, len(snapshot.entries))
	for _, entry := range snapshot.entries {
		plan = append(plan, entry.advance)
	}
	sourceOptions, ok := packs.OperationPackSourceOptions(ctx)
	if !ok {
		sourceOptions = packs.DefaultPackSourceOptions()
	}
	resolution, err := resolveApplyAllSourceRevisionAdvanceSourcesAtRevisionsWithBase(
		plan, snapshot.replayRevisions(), r.componentKindRegistrationSet(), sourceOptions,
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceTuple{}, err
	}
	replay, err := resolveApplyAllSourceRevisionAdvanceInstalledTupleFromSnapshot(
		ctx, r, snapshot, resolution, activeRuntime, nil,
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceTuple{}, err
	}
	precondition := applyAllSourceRevisionAdvancePreconditionProof{
		snapshot: snapshot, activeRuntime: activeRuntime, current: current, replay: replay,
	}
	if err := validateApplyAllSourceRevisionAdvanceReplayTuple(precondition); err != nil {
		return applyAllSourceRevisionAdvanceTuple{}, err
	}
	return replay, nil
}

func cloneApplyAllSourceRevisionAdvanceTuple(tuple applyAllSourceRevisionAdvanceTuple) applyAllSourceRevisionAdvanceTuple {
	out := applyAllSourceRevisionAdvanceTuple{
		plan: append([]normalizedApplyAllSourceRevisionAdvance(nil), tuple.plan...),
	}
	if tuple.targets != nil {
		out.targets = make(map[string]*packs.SourceBackedPack, len(tuple.targets))
		for name, target := range tuple.targets {
			out.targets[name] = target
		}
	}
	return out
}

// validateApplyAllSourceRevisionAdvanceReplayTuple checks only construction
// and endpoint facts. The later generation-authority preflight is the canonical
// proof that a retry replay base equals the still-active immutable generation.
func validateApplyAllSourceRevisionAdvanceReplayTuple(
	precondition applyAllSourceRevisionAdvancePreconditionProof,
) error {
	if precondition.replay.plan == nil || precondition.replay.targets == nil {
		return applyAllSourceRevisionAdvancePlanError(
			ApplyAllSourceRevisionAdvancePlanState, -1, "", fmt.Errorf("replay tuple was not minted"),
		)
	}
	if len(precondition.snapshot.entries) != len(precondition.replay.plan) ||
		len(precondition.snapshot.entries) != len(precondition.replay.targets) ||
		len(precondition.snapshot.entries) != len(precondition.current.plan) {
		return applyAllSourceRevisionAdvancePlanError(
			ApplyAllSourceRevisionAdvancePlanState, -1, "", fmt.Errorf("replay tuple does not match the frozen precondition cardinality"),
		)
	}
	for index, entry := range precondition.snapshot.entries {
		if precondition.replay.plan[index] != entry.advance {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState, entry.advance.index, entry.advance.Name,
				fmt.Errorf("replay tuple plan does not match the frozen precondition"),
			)
		}
	}
	seen := make(map[string]struct{}, len(precondition.snapshot.entries))
	for _, entry := range precondition.snapshot.entries {
		advance := entry.advance
		name := strings.ToLower(strings.TrimSpace(advance.Name))
		if name == "" {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState, advance.index, advance.Name, fmt.Errorf("replay tuple pack name is empty"),
			)
		}
		if _, duplicate := seen[name]; duplicate {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState, advance.index, advance.Name, fmt.Errorf("replay tuple duplicates pack name %q", name),
			)
		}
		seen[name] = struct{}{}
		target := precondition.replay.targets[name]
		if target == nil {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState, advance.index, advance.Name, fmt.Errorf("replay tuple target is missing"),
			)
		}
		expected := advance
		if entry.pending != nil && entry.pending.Kind == ApplyAllSourceRevisionAdvancePendingRetry {
			expected.ToRevision = advance.FromRevision
		} else {
			expected.ToRevision = strings.TrimSpace(entry.row.SourceRevision.String)
		}
		if err := validateApplyAllSourceRevisionAdvanceTarget(expected, target); err != nil {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState, advance.index, advance.Name,
				fmt.Errorf("validate replay tuple target: %w", err),
			)
		}
	}
	if len(seen) != len(precondition.replay.targets) {
		return applyAllSourceRevisionAdvancePlanError(
			ApplyAllSourceRevisionAdvancePlanState, -1, "", fmt.Errorf("replay tuple has targets outside the frozen plan"),
		)
	}
	return nil
}