package controlplane

import (
	"context"
	"errors"
	"fmt"
	"sync/atomic"

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

type applyAllSourceRevisionAdvanceFinalizationClaim struct {
	claimed atomic.Bool
}

// applyAllSourceRevisionAdvanceExecution is the opaque proof that the one atom
// selected by Begin executed successfully against the exact registered runtime.
// Exact finalization consumes its separate claim; the carrier exposes no
// lifecycle transition by itself.
type applyAllSourceRevisionAdvanceExecution struct {
	identity *applyAllSourceRevisionAdvanceExecutionIdentity
	witness  applyAllSourceRevisionAdvanceExecutionProvenance
}

type applyAllSourceRevisionAdvanceExecutionIdentity struct {
	registration      applyAllSourceRevisionAdvanceRuntimeRegistration
	selectedAtomClaim *applyAllSourceRevisionAdvanceSelectedAtomClaim
	finalizationClaim *applyAllSourceRevisionAdvanceFinalizationClaim
	provenance        *applyAllSourceRevisionAdvanceExecutionProvenance
	minted            bool
}

type applyAllSourceRevisionAdvanceExecutionProvenance struct {
	registrationIdentity *applyAllSourceRevisionAdvanceRuntimeRegistrationIdentity
	selectedAtomClaim    *applyAllSourceRevisionAdvanceSelectedAtomClaim
	finalizationClaim    *applyAllSourceRevisionAdvanceFinalizationClaim
}

type applyAllSourceRevisionAdvanceExecutionState struct {
	registration      applyAllSourceRevisionAdvanceRuntimeRegistrationState
	finalizationClaim *applyAllSourceRevisionAdvanceFinalizationClaim
}

type applyAllSourceRevisionAdvanceExecutionOperations struct {
	execute func(
		context.Context,
		*packs.SourceBackedExecutionAtom,
		packs.ApplyDeps,
		packs.InstallPlanHooks,
	) error
	compensate func(
		context.Context,
		applyAllSourceRevisionAdvanceRuntimeRegistration,
		error,
	) error
}

// executeApplyAllSourceRevisionAdvanceSelectedAtom consumes the Begin-minted
// selected-atom claim exactly once. Every error returned by the atom includes
// managed-component preflight failures, component failures, and AfterPhase
// failures, and therefore enters the same mandatory compensation path.
func executeApplyAllSourceRevisionAdvanceSelectedAtom(
	ctx context.Context,
	registration applyAllSourceRevisionAdvanceRuntimeRegistration,
) (applyAllSourceRevisionAdvanceExecution, error) {
	return executeApplyAllSourceRevisionAdvanceSelectedAtomWith(
		ctx,
		registration,
		applyAllSourceRevisionAdvanceExecutionOperations{
			execute: func(
				ctx context.Context,
				atom *packs.SourceBackedExecutionAtom,
				deps packs.ApplyDeps,
				hooks packs.InstallPlanHooks,
			) error {
				return atom.Execute(ctx, deps, hooks)
			},
			compensate: compensateApplyAllSourceRevisionAdvanceExecution,
		},
	)
}

func executeApplyAllSourceRevisionAdvanceSelectedAtomWith(
	ctx context.Context,
	registration applyAllSourceRevisionAdvanceRuntimeRegistration,
	operations applyAllSourceRevisionAdvanceExecutionOperations,
) (applyAllSourceRevisionAdvanceExecution, error) {
	state, err := requireApplyAllSourceRevisionAdvanceRuntimeRegistration(registration)
	if err != nil {
		return applyAllSourceRevisionAdvanceExecution{}, err
	}
	attempt := state.snapshots.attempt
	advance := attempt.advance
	if operations.execute == nil || operations.compensate == nil {
		return applyAllSourceRevisionAdvanceExecution{}, applyAllSourceRevisionAdvanceAttemptError(
			"execute", advance, fmt.Errorf("execution operations are incomplete"),
		)
	}

	claim := registration.identity.selectedAtomClaim
	if !claim.claimed.CompareAndSwap(false, true) {
		return applyAllSourceRevisionAdvanceExecution{}, applyAllSourceRevisionAdvanceAttemptError(
			"execute", advance, fmt.Errorf("selected execution atom was already claimed"),
		)
	}

	proof := attempt.identity.admission.identity.proof
	entry := proof.entries[attempt.identity.selectedProofIndex]
	runtime := proof.executionRuntime
	deps := runtime.applyDeps
	deps.Options = attempt.identity.applyOptions
	hooks := packs.InstallPlanHooks{
		AfterPhase: func(ctx context.Context, phase packs.InstallPlanPhase) error {
			return runSourceBackedInstallPhaseBoundaryWithRuntime(
				ctx,
				runtime.runnerDatabase,
				runtime.serviceGraph,
				attempt.target,
				phase,
				"apply-all source revision advance",
			)
		},
	}
	if err := operations.execute(ctx, entry.atom, deps, hooks); err != nil {
		return applyAllSourceRevisionAdvanceExecution{}, compensateApplyAllSourceRevisionAdvanceOperationError(
			ctx, registration, advance, "execute", err, operations.compensate,
		)
	}

	identity := &applyAllSourceRevisionAdvanceExecutionIdentity{
		registration:      registration,
		selectedAtomClaim: claim,
		finalizationClaim: &applyAllSourceRevisionAdvanceFinalizationClaim{},
		minted:            true,
	}
	identity.provenance = &applyAllSourceRevisionAdvanceExecutionProvenance{
		registrationIdentity: registration.identity,
		selectedAtomClaim:    claim,
		finalizationClaim:    identity.finalizationClaim,
	}
	execution := applyAllSourceRevisionAdvanceExecution{
		identity: identity,
		witness:  *identity.provenance,
	}
	if _, err := requireApplyAllSourceRevisionAdvanceExecution(execution); err != nil {
		return applyAllSourceRevisionAdvanceExecution{}, compensateApplyAllSourceRevisionAdvanceOperationError(
			ctx, registration, advance, "execute", err, operations.compensate,
		)
	}
	return execution, nil
}

func compensateApplyAllSourceRevisionAdvanceOperationError(
	ctx context.Context,
	registration applyAllSourceRevisionAdvanceRuntimeRegistration,
	advance normalizedApplyAllSourceRevisionAdvance,
	operation string,
	operationErr error,
	compensate func(
		context.Context,
		applyAllSourceRevisionAdvanceRuntimeRegistration,
		error,
	) error,
) error {
	operationErr = applyAllSourceRevisionAdvanceAttemptError(operation, advance, operationErr)
	recoveryErr := compensate(ctx, registration, operationErr)
	if recoveryErr == nil {
		return operationErr
	}
	if errors.Is(recoveryErr, operationErr) {
		return recoveryErr
	}
	return errors.Join(operationErr, recoveryErr)
}

func requireApplyAllSourceRevisionAdvanceExecution(
	execution applyAllSourceRevisionAdvanceExecution,
) (applyAllSourceRevisionAdvanceExecutionState, error) {
	identity := execution.identity
	if identity == nil || !identity.minted || identity.provenance == nil {
		return applyAllSourceRevisionAdvanceExecutionState{}, fmt.Errorf(
			"selected-atom execution was not minted by its execution boundary",
		)
	}
	if identity.registration.identity == nil ||
		identity.registration.identity != identity.provenance.registrationIdentity ||
		identity.registration.identity != execution.witness.registrationIdentity {
		return applyAllSourceRevisionAdvanceExecutionState{}, fmt.Errorf(
			"selected-atom execution changed runtime registration identity",
		)
	}
	if identity.selectedAtomClaim == nil ||
		identity.selectedAtomClaim != identity.registration.identity.selectedAtomClaim ||
		identity.selectedAtomClaim != identity.provenance.selectedAtomClaim ||
		identity.selectedAtomClaim != execution.witness.selectedAtomClaim {
		return applyAllSourceRevisionAdvanceExecutionState{}, fmt.Errorf(
			"selected-atom execution changed its one-shot claim",
		)
	}
	if identity.finalizationClaim == nil ||
		identity.finalizationClaim != identity.provenance.finalizationClaim ||
		identity.finalizationClaim != execution.witness.finalizationClaim {
		return applyAllSourceRevisionAdvanceExecutionState{}, fmt.Errorf(
			"selected-atom execution changed its finalization claim",
		)
	}
	if !identity.selectedAtomClaim.claimed.Load() {
		return applyAllSourceRevisionAdvanceExecutionState{}, fmt.Errorf(
			"selected-atom execution claim was not consumed",
		)
	}
	registrationState, err := requireApplyAllSourceRevisionAdvanceRuntimeRegistration(
		identity.registration,
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceExecutionState{}, fmt.Errorf(
			"validate selected-atom execution registration: %w", err,
		)
	}
	return applyAllSourceRevisionAdvanceExecutionState{
		registration:      registrationState,
		finalizationClaim: identity.finalizationClaim,
	}, nil
}