package clusterrecovery

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strings"
	"time"
	"unicode"

	"github.com/pingcap-inc/clinic/pkg/db"
)

var (
	ErrApplyGuard  = errors.New("recovery apply guard rejected the request")
	ErrSourceDrift = errors.New("authoritative source changed after recovery planning")
)

type ApplyOptions struct {
	AllowWrite               bool
	ExpectedManifestHash     string
	ApprovalRef              string
	WorkloadIdentity         string
	MutationClass            MutationClass
	Phase                    ApplyPhase
	Offset                   int
	Limit                    int
	PriorReceipt             *Receipt
	ExpectedPriorReceiptHash string
	PriorVerificationRef     string
}

type Executor struct {
	source Source
	store  RecoveryStore
	now    func() time.Time
}

func NewExecutor(source Source, store RecoveryStore) *Executor {
	return &Executor{source: source, store: store, now: time.Now}
}

// ValidateApplyRequest verifies every non-I/O apply gate. Commands should call
// this before initializing external dependencies; Execute repeats the check so a
// caller cannot bypass it.
func ValidateApplyRequest(manifest *Manifest, options ApplyOptions, now time.Time) error {
	_, _, _, err := validateApplyOptions(manifest, options, now.UTC())
	return err
}

type preparedAction struct {
	index      int
	action     Action
	desired    *db.Cluster
	before     *db.Cluster
	beforeHash string
	idempotent bool
}

func (e *Executor) Execute(
	ctx context.Context,
	manifest *Manifest,
	options ApplyOptions,
	rollbackSink RollbackSink,
) (*Receipt, error) {
	if e == nil || e.source == nil || e.store == nil {
		return nil, errors.New("executor source and store are required")
	}
	now := e.now().UTC()
	approvalHash, priorHash, verificationHash, err := validateApplyOptions(manifest, options, now)
	if err != nil {
		return nil, err
	}
	workloadIdentityHash, err := checksumJSON(options.WorkloadIdentity)
	if err != nil {
		return nil, err
	}
	if rollbackSink == nil {
		return nil, fmt.Errorf("%w: rollback sink is required", ErrApplyGuard)
	}
	receipt := &Receipt{
		SchemaVersion:        SchemaVersion,
		Environment:          StagingEnvironment,
		RunID:                manifest.RunID,
		ManifestHash:         manifest.ManifestHash,
		MutationClass:        manifest.MutationClass,
		Phase:                options.Phase,
		Offset:               options.Offset,
		Limit:                options.Limit,
		ApprovalRefHash:      approvalHash,
		WorkloadIdentityHash: workloadIdentityHash,
		PriorReceiptHash:     priorHash,
		VerificationRefHash:  verificationHash,
		StartedAt:            now,
	}

	selected := manifest.Actions[options.Offset : options.Offset+options.Limit]
	prepared := make([]preparedAction, 0, len(selected))
	rollbackEntries := make([]RollbackEntry, 0, len(selected))
	availableAfterApply := make(map[string]struct{}, len(selected))
	for relativeIndex, action := range selected {
		index := options.Offset + relativeIndex
		item, err := e.prepareAction(ctx, manifest.MutationClass, index, action, availableAfterApply)
		if err != nil {
			receipt.Stopped = true
			receipt.Results = append(receipt.Results, ExecutionResult{
				ActionIndex: index,
				Kind:        action.Kind,
				ClusterID:   action.ClusterID,
				Outcome:     MutationOutcomeFailed,
				ReasonCode:  executionReasonCode(err),
			})
			return finalizeStoppedReceipt(receipt, e.now().UTC(), fmt.Errorf("recovery preflight: %w", err))
		}
		prepared = append(prepared, *item)
		availableAfterApply[action.ClusterID] = struct{}{}
		if item.idempotent {
			continue
		}
		entry := RollbackEntry{
			ActionIndex:             index,
			Kind:                    action.Kind,
			ClusterID:               action.ClusterID,
			ExpectedDesiredChecksum: action.DesiredChecksum,
		}
		if manifest.MutationClass == MutationCreate {
			entry.Operation = RollbackDeleteCreated
		} else {
			entry.Operation = RollbackRestoreUpdate
			entry.BeforeChecksum = item.beforeHash
			entry.Before = persistentClusterFrom(item.before)
		}
		rollbackEntries = append(rollbackEntries, entry)
	}

	rollbackBundle := &RollbackBundle{
		SchemaVersion: SchemaVersion,
		Environment:   StagingEnvironment,
		RunID:         manifest.RunID,
		ManifestHash:  manifest.ManifestHash,
		MutationClass: manifest.MutationClass,
		GeneratedAt:   e.now().UTC(),
		Entries:       rollbackEntries,
	}
	if err := FinalizeRollbackBundle(rollbackBundle); err != nil {
		return finalizeStoppedReceipt(receipt, e.now().UTC(), fmt.Errorf("finalize rollback evidence: %w", err))
	}
	receipt.RollbackBundleHash = rollbackBundle.BundleHash
	if err := rollbackSink.WriteRollbackBundle(rollbackBundle); err != nil {
		return finalizeStoppedReceipt(receipt, e.now().UTC(), fmt.Errorf("persist rollback evidence before write: %w", err))
	}

	for _, item := range prepared {
		if err := ctx.Err(); err != nil {
			receipt.Stopped = true
			receipt.Results = append(receipt.Results, failedExecutionResult(item, "context_canceled"))
			return finalizeStoppedReceipt(receipt, e.now().UTC(), err)
		}
		fresh, err := e.prepareAction(ctx, manifest.MutationClass, item.index, item.action, nil)
		if err != nil {
			receipt.Stopped = true
			receipt.Results = append(receipt.Results, failedExecutionResult(item, executionReasonCode(err)))
			return finalizeStoppedReceipt(receipt, e.now().UTC(), fmt.Errorf("immediate recovery revalidation: %w", err))
		}
		if item.idempotent != fresh.idempotent {
			receipt.Stopped = true
			receipt.Results = append(receipt.Results, failedExecutionResult(item, "row_drift"))
			return finalizeStoppedReceipt(receipt, e.now().UTC(), fmt.Errorf("immediate recovery revalidation: %w", ErrRowDrift))
		}
		item = *fresh
		if item.idempotent {
			receipt.Results = append(receipt.Results, ExecutionResult{
				ActionIndex:    item.index,
				Kind:           item.action.Kind,
				ClusterID:      item.action.ClusterID,
				Outcome:        MutationOutcomeNoop,
				BeforeChecksum: item.beforeHash,
				AfterChecksum:  item.beforeHash,
			})
			continue
		}
		var mutation *MutationResult
		var mutationErr error
		if manifest.MutationClass == MutationCreate {
			mutation, mutationErr = e.store.ApplyCreate(ctx, item.action.Kind, item.desired, item.action.DesiredChecksum)
		} else {
			mutation, mutationErr = e.store.ApplyUpdate(
				ctx,
				item.action.Kind,
				item.desired,
				item.action.DesiredChecksum,
				item.action.BeforeChecksum,
				item.action.Fields,
			)
		}
		if mutationErr != nil || mutation == nil {
			receipt.Stopped = true
			reason := executionReasonCode(mutationErr)
			if mutation == nil && mutationErr == nil {
				mutationErr = errors.New("mutation result is missing")
				reason = "mutation_result_missing"
			}
			receipt.Results = append(receipt.Results, failedExecutionResult(item, reason))
			return finalizeStoppedReceipt(receipt, e.now().UTC(), fmt.Errorf("execute guarded mutation: %w", mutationErr))
		}
		if err := validateMutationResult(manifest.MutationClass, mutation); err != nil {
			receipt.Stopped = true
			receipt.Results = append(receipt.Results, failedExecutionResult(item, "mutation_result_invalid"))
			return finalizeStoppedReceipt(receipt, e.now().UTC(), err)
		}
		receipt.Results = append(receipt.Results, ExecutionResult{
			ActionIndex:    item.index,
			Kind:           item.action.Kind,
			ClusterID:      item.action.ClusterID,
			Outcome:        mutation.Outcome,
			BeforeChecksum: mutation.BeforeChecksum,
			AfterChecksum:  mutation.AfterChecksum,
		})
	}
	receipt.CompletedAt = e.now().UTC()
	if err := FinalizeReceipt(receipt); err != nil {
		return nil, err
	}
	return receipt, nil
}

func validateMutationResult(mutationClass MutationClass, result *MutationResult) error {
	if result == nil || (result.Outcome != MutationOutcomeApplied && result.Outcome != MutationOutcomeNoop) ||
		!isSHA256(result.AfterChecksum) {
		return errors.New("guarded mutation returned invalid evidence")
	}
	if result.Outcome == MutationOutcomeNoop {
		if !isSHA256(result.BeforeChecksum) || result.BeforeChecksum != result.AfterChecksum {
			return errors.New("guarded no-op returned invalid evidence")
		}
		return nil
	}
	if mutationClass == MutationCreate {
		if result.BeforeChecksum != "" {
			return errors.New("guarded create returned an unexpected before checksum")
		}
		return nil
	}
	if mutationClass != MutationUpdate || !isSHA256(result.BeforeChecksum) || result.BeforeChecksum == result.AfterChecksum {
		return errors.New("guarded update returned invalid evidence")
	}
	return nil
}

func (e *Executor) prepareAction(
	ctx context.Context,
	mutationClass MutationClass,
	index int,
	action Action,
	availableAfterApply map[string]struct{},
) (*preparedAction, error) {
	candidate := Candidate{
		Kind:        action.Kind,
		ClusterID:   action.ClusterID,
		TenantID:    action.TenantID,
		ClusterType: action.ClusterType,
	}
	resolved, err := e.source.Resolve(ctx, candidate)
	if err != nil || resolved == nil || resolved.Terminal != action.Terminal || resolved.Desired == nil {
		return nil, ErrSourceDrift
	}
	if !resolved.SourceUpdatedAt.UTC().Equal(action.SourceUpdatedAt.UTC()) {
		return nil, ErrSourceDrift
	}
	expectedDependencies := make([]string, 0, 1)
	if resolved.Parent != nil {
		expectedDependencies = append(expectedDependencies, resolved.Parent.ClusterID)
	}
	if !slices.Equal(expectedDependencies, action.DependencyIDs) {
		return nil, ErrSourceDrift
	}
	snapshot, err := e.store.ReadSnapshot(ctx, action.Kind, action.TenantID, action.ClusterID)
	if err != nil || snapshot == nil || snapshot.Organization == nil {
		return nil, ErrRowDrift
	}
	if snapshot.Cluster != nil && (snapshot.Cluster.TenantID != action.TenantID || snapshot.Cluster.OrgID != snapshot.Organization.ID) {
		return nil, ErrRowDrift
	}
	desired := *resolved.Desired
	desired.OrgID = snapshot.Organization.ID
	if desired.ClusterID != action.ClusterID || desired.TenantID != action.TenantID ||
		terminalStatus(desired.Status) != action.Terminal || validateDesiredLifecycle(&desired) != nil {
		return nil, ErrSourceDrift
	}
	desiredChecksum, err := desiredClusterChecksum(action.Kind, &desired)
	if err != nil || desiredChecksum != action.DesiredChecksum {
		return nil, ErrSourceDrift
	}
	for _, dependencyID := range action.DependencyIDs {
		if _, available := availableAfterApply[dependencyID]; available {
			continue
		}
		dependency, err := e.store.ReadSnapshot(ctx, ObjectKindResourcePool, action.TenantID, dependencyID)
		if err != nil || dependency == nil || dependency.Organization == nil || dependency.Cluster == nil ||
			dependency.Cluster.TenantID != action.TenantID || dependency.Cluster.OrgID != snapshot.Organization.ID ||
			dependency.Cluster.DeletedAt.Valid || terminalStatus(dependency.Cluster.Status) {
			return nil, ErrParentUnavailable
		}
	}

	prepared := &preparedAction{index: index, action: action, desired: &desired, before: snapshot.Cluster}
	if snapshot.Cluster != nil {
		prepared.beforeHash, err = clusterRecordChecksum(snapshot.Cluster)
		if err != nil {
			return nil, ErrRowDrift
		}
	}
	switch mutationClass {
	case MutationCreate:
		if snapshot.Cluster == nil {
			return prepared, nil
		}
		if err := verifyExistingIdentity(snapshot.Cluster, &desired); err != nil {
			return nil, ErrRowDrift
		}
		changes, err := diffOwnedFields(action.Kind, snapshot.Cluster, &desired)
		if err != nil || len(changes) != 0 {
			return nil, ErrRowDrift
		}
		prepared.idempotent = true
		return prepared, nil
	case MutationUpdate:
		if snapshot.Cluster == nil || (hasTerminalMarker(snapshot.Cluster) && !action.Terminal) {
			return nil, ErrRowDrift
		}
		if err := verifyExistingIdentity(snapshot.Cluster, &desired); err != nil {
			return nil, ErrRowDrift
		}
		changes, err := diffOwnedFields(action.Kind, snapshot.Cluster, &desired)
		if err != nil {
			return nil, ErrRowDrift
		}
		if len(changes) == 0 {
			prepared.idempotent = true
			return prepared, nil
		}
		if prepared.beforeHash != action.BeforeChecksum || !slices.Equal(changeFieldNames(changes), action.Fields) {
			return nil, ErrRowDrift
		}
		return prepared, nil
	default:
		return nil, ErrApplyGuard
	}
}

func validateApplyOptions(manifest *Manifest, options ApplyOptions, now time.Time) (string, string, string, error) {
	if !options.AllowWrite {
		return "", "", "", fmt.Errorf("%w: --allow-write is required", ErrApplyGuard)
	}
	if err := VerifyManifest(manifest, options.ExpectedManifestHash, now); err != nil {
		return "", "", "", fmt.Errorf("%w: manifest verification failed", ErrApplyGuard)
	}
	if options.MutationClass != manifest.MutationClass {
		return "", "", "", fmt.Errorf("%w: mutation class mismatch", ErrApplyGuard)
	}
	if err := validateReference(options.ApprovalRef); err != nil {
		return "", "", "", fmt.Errorf("%w: invalid approval reference", ErrApplyGuard)
	}
	if err := validateReference(options.WorkloadIdentity); err != nil {
		return "", "", "", fmt.Errorf("%w: invalid workload identity", ErrApplyGuard)
	}
	if options.Offset < 0 || options.Limit <= 0 || options.Offset+options.Limit > len(manifest.Actions) {
		return "", "", "", fmt.Errorf("%w: action bounds are invalid", ErrApplyGuard)
	}
	if options.Phase == ApplyPhaseCanary {
		if options.Offset != 0 || options.Limit > 3 || options.PriorReceipt != nil ||
			options.ExpectedPriorReceiptHash != "" || options.PriorVerificationRef != "" {
			return "", "", "", fmt.Errorf("%w: Canary must start at zero and contain at most three actions", ErrApplyGuard)
		}
	} else if options.Phase == ApplyPhaseBatch {
		if options.Limit > 10 || options.PriorReceipt == nil ||
			options.Offset != options.PriorReceipt.Offset+len(options.PriorReceipt.Results) {
			return "", "", "", fmt.Errorf("%w: batch bounds or prior receipt are invalid", ErrApplyGuard)
		}
		if err := VerifyReceipt(options.PriorReceipt, options.ExpectedPriorReceiptHash); err != nil ||
			!successfulPriorReceipt(options.PriorReceipt, manifest) || validateReference(options.PriorVerificationRef) != nil {
			return "", "", "", fmt.Errorf("%w: verified prior-batch evidence is required", ErrApplyGuard)
		}
	} else {
		return "", "", "", fmt.Errorf("%w: apply phase is invalid", ErrApplyGuard)
	}
	approvalHash, err := checksumJSON(options.ApprovalRef)
	if err != nil {
		return "", "", "", err
	}
	priorHash := ""
	verificationHash := ""
	if options.Phase == ApplyPhaseBatch {
		priorHash = options.ExpectedPriorReceiptHash
		verificationHash, err = checksumJSON(options.PriorVerificationRef)
		if err != nil {
			return "", "", "", err
		}
	}
	return approvalHash, priorHash, verificationHash, nil
}

func validateReference(value string) error {
	if value == "" || strings.TrimSpace(value) != value || len(value) > 256 {
		return errors.New("reference must be 1-256 trimmed characters")
	}
	for _, character := range value {
		if unicode.IsControl(character) {
			return errors.New("reference contains a control character")
		}
	}
	return nil
}

func executionReasonCode(err error) string {
	switch {
	case errors.Is(err, ErrSourceDrift):
		return "source_drift"
	case errors.Is(err, ErrParentUnavailable):
		return "parent_unavailable"
	case errors.Is(err, ErrRowDrift):
		return "row_drift"
	case errors.Is(err, ErrMutationConflict):
		return "mutation_conflict"
	case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
		return "context_canceled"
	default:
		return "mutation_error"
	}
}

func failedExecutionResult(item preparedAction, reason string) ExecutionResult {
	return ExecutionResult{
		ActionIndex:    item.index,
		Kind:           item.action.Kind,
		ClusterID:      item.action.ClusterID,
		Outcome:        MutationOutcomeFailed,
		ReasonCode:     reason,
		BeforeChecksum: item.beforeHash,
	}
}

func finalizeStoppedReceipt(receipt *Receipt, completedAt time.Time, cause error) (*Receipt, error) {
	receipt.Stopped = true
	receipt.CompletedAt = completedAt.UTC()
	if err := FinalizeReceipt(receipt); err != nil {
		return nil, errors.Join(cause, err)
	}
	return receipt, cause
}