package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"sync/atomic"

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

// applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt is an opaque,
// one-shot witness that one exact prepared target passed every live preflight
// before Begin. It exposes no direct mutation or execution method.
type applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt struct {
	identity *applyAllSourceRevisionAdvanceSelectedLivePreflightIdentity
	witness  applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance
}

// applyAllSourceRevisionAdvanceSelectedLivePreflightClaim is the single
// one-shot lifecycle cell shared by the receipt identity and both detached
// witnesses. It owns no target, database, selection, or execution fact.
type applyAllSourceRevisionAdvanceSelectedLivePreflightClaim struct {
	claimed atomic.Bool
}

type applyAllSourceRevisionAdvanceSelectedLivePreflightIdentity struct {
	prepared   *applyAllSourceRevisionAdvanceAttemptIdentity
	finalRow   installedPackRow
	claim      *applyAllSourceRevisionAdvanceSelectedLivePreflightClaim
	provenance *applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance
	minted     bool
}

// applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance retains a
// detached attempt witness captured before any live read. Target, database,
// registrations, dependencies, selection, and options remain derived from that
// one prepared authority rather than becoming competing receipt facts.
type applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance struct {
	prepared        *applyAllSourceRevisionAdvanceAttemptIdentity
	preparedWitness applyAllSourceRevisionAdvanceAttemptProvenance
	finalRow        installedPackRow
	claim           *applyAllSourceRevisionAdvanceSelectedLivePreflightClaim
	minted          bool
}

type applyAllSourceRevisionAdvanceSelectedLivePreflightState struct {
	prepared       *applyAllSourceRevisionAdvanceAttemptIdentity
	advance        normalizedApplyAllSourceRevisionAdvance
	target         *packs.SourceBackedPack
	database       *sql.DB
	runnerDatabase *database.TiDBConnection
	applyDeps      packs.ApplyDeps
	registrations  packs.ComponentKindRegistrationSet
	applyOptions   packs.ApplyOptions
}

// preflightApplyAllSourceRevisionAdvanceSelectedTarget performs the complete
// point-in-time live validation for one already-prepared advance or retry. It
// must eventually be called and consumed under the same control-plane advisory
// lock; the ApplyAll orchestrator retains that scope through Begin and terminal
// reconciliation. The Begin boundary consumes only this receipt.
func preflightApplyAllSourceRevisionAdvanceSelectedTarget(
	ctx context.Context,
	prepared *applyAllSourceRevisionAdvanceAttemptIdentity,
	offlineCandidate *offlineCandidateExecution,
) (applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt, error) {
	selectedState, err := requireApplyAllSourceRevisionAdvancePreparedAttributionState(prepared)
	if err != nil {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt{}, err
	}
	selected := selectedState.selected
	advance := selected.current.advance
	runtime := selectedState.proof.executionRuntime
	preparedWitness := cloneApplyAllSourceRevisionAdvanceAttemptProvenance(prepared.provenance)
	fail := func(err error) (applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt, error) {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt{},
			applyAllSourceRevisionAdvanceAttemptError("preflight", advance, err)
	}

	if err := ensureInstallPackSchemaReadinessForConnection(runtime.runnerDatabase); err != nil {
		return fail(err)
	}
	overrides, err := buildApplyAllSourceRevisionAdvanceAttributionPreflightOverrides(prepared)
	if err != nil {
		return fail(err)
	}
	overrides.offlineCandidatePrepared = offlineCandidate
	if err := validateLivePackDependenciesWithPreflightOverrides(
		ctx,
		runtime.database,
		selected.installedCompatibility,
		overrides,
	); err != nil {
		return fail(fmt.Errorf("validate exact desired dependencies: %w", err))
	}
	applyDeps := runtime.applyDeps
	applyDeps.Options = prepared.applyOptions
	if err := validateSourceBackedPackInstallPreflights(
		ctx,
		applyDeps,
		runtime.componentKindRegistrationSet,
		selected.installedCompatibility,
		overrides,
	); err != nil {
		return fail(err)
	}

	hasRuntimeDeclarations, err := installedPackHasRuntimeDeclarationColumns(ctx, runtime.database)
	if err != nil {
		return fail(err)
	}
	if !hasRuntimeDeclarations {
		return fail(fmt.Errorf("installed-pack schema is missing runtime declaration columns"))
	}
	hasRuntimeStartupShallow, err := installedPackHasRuntimeStartupShallowDeclarationColumn(ctx, runtime.database)
	if err != nil {
		return fail(err)
	}
	if !hasRuntimeStartupShallow {
		return fail(fmt.Errorf("installed-pack schema is missing runtime startup shallow declaration column"))
	}
	hasRuntimeFormulaBundles, err := installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx, runtime.database)
	if err != nil {
		return fail(err)
	}
	if !hasRuntimeFormulaBundles {
		return fail(fmt.Errorf("installed-pack schema is missing runtime formula function bundles column"))
	}
	hasHostRequirementsDigest, err := installedPackHasHostRequirementsDigestColumn(ctx, runtime.database)
	if err != nil {
		return fail(err)
	}
	if !hasHostRequirementsDigest {
		return fail(fmt.Errorf("installed-pack schema is missing host requirements digest column"))
	}

	finalRow, ok, err := getInstalledPackRowByNameForSourceRevisionAdvance(
		ctx, runtime.database, advance.Name,
	)
	if err != nil {
		return fail(fmt.Errorf("final installed-pack row reread: %w", err))
	}
	if !ok {
		return fail(fmt.Errorf("installed pack row disappeared after live preflight"))
	}
	if finalRow != selected.current.row {
		return fail(fmt.Errorf("installed pack row changed during live preflight"))
	}
	pending, err := classifyApplyAllSourceRevisionAdvance(
		finalRow, advance.SourceRevisionAdvanceAuthorization,
	)
	if err != nil {
		return fail(fmt.Errorf("reclassify final installed-pack row: %w", err))
	}
	if !equalApplyAllSourceRevisionAdvancePending(pending, selected.current.pending) {
		return fail(fmt.Errorf("pending source revision advance changed during live preflight"))
	}

	witnessedPrepared := *prepared
	witnessedPrepared.provenance = cloneApplyAllSourceRevisionAdvanceAttemptProvenance(preparedWitness)
	if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(&witnessedPrepared, false); err != nil {
		return fail(fmt.Errorf("prepared authority changed during live preflight: %w", err))
	}
	claim := &applyAllSourceRevisionAdvanceSelectedLivePreflightClaim{}
	provenance := &applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance{
		prepared: prepared, preparedWitness: preparedWitness,
		finalRow: finalRow, claim: claim, minted: true,
	}
	identity := &applyAllSourceRevisionAdvanceSelectedLivePreflightIdentity{
		prepared: prepared, finalRow: finalRow,
		claim: claim, provenance: provenance, minted: true,
	}
	receipt := applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt{
		identity: identity,
		witness:  cloneApplyAllSourceRevisionAdvanceSelectedLivePreflightProvenance(*provenance),
	}
	if _, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt); err != nil {
		return fail(err)
	}
	return receipt, nil
}

func requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(
	receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt,
) (applyAllSourceRevisionAdvanceSelectedLivePreflightState, error) {
	identity := receipt.identity
	if identity == nil || !identity.minted || identity.provenance == nil ||
		!identity.provenance.minted || !receipt.witness.minted {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"selected live-preflight receipt was not minted by its preflight boundary",
		)
	}
	provenance := identity.provenance
	witness := receipt.witness
	if identity.prepared == nil || identity.prepared != provenance.prepared ||
		identity.prepared != witness.prepared {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"selected live-preflight receipt changed prepared-attempt identity",
		)
	}
	if identity.finalRow != provenance.finalRow || identity.finalRow != witness.finalRow {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"selected live-preflight receipt changed its final installed row",
		)
	}
	if identity.claim == nil || identity.claim != provenance.claim || identity.claim != witness.claim {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"selected live-preflight receipt changed its one-shot claim identity",
		)
	}
	if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(identity.prepared, false); err != nil {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"validate selected live-preflight prepared authority: %w", err,
		)
	}
	for _, preparedWitness := range []applyAllSourceRevisionAdvanceAttemptProvenance{
		provenance.preparedWitness,
		witness.preparedWitness,
	} {
		witnessedPrepared := *identity.prepared
		witnessedPrepared.provenance = cloneApplyAllSourceRevisionAdvanceAttemptProvenance(preparedWitness)
		if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(&witnessedPrepared, false); err != nil {
			return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
				"validate selected live-preflight detached prepared authority: %w", err,
			)
		}
	}
	selectedState, err := requireApplyAllSourceRevisionAdvancePreparedAttributionState(identity.prepared)
	if err != nil {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, err
	}
	selected := selectedState.selected
	runtime := selectedState.proof.executionRuntime
	if identity.finalRow != selected.current.row {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"selected live-preflight final row does not match the prepared target",
		)
	}
	applyDeps := runtime.applyDeps
	applyDeps.Options = identity.prepared.applyOptions
	return applyAllSourceRevisionAdvanceSelectedLivePreflightState{
		prepared:       identity.prepared,
		advance:        selected.current.advance,
		target:         selected.installedCompatibility,
		database:       runtime.database,
		runnerDatabase: runtime.runnerDatabase,
		applyDeps:      applyDeps,
		registrations:  runtime.componentKindRegistrationSet,
		applyOptions:   identity.prepared.applyOptions,
	}, nil
}

// claimApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt is the only
// one-shot transition from immutable live-preflight evidence to Begin. The
// receipt remains claimed when Begin later fails, so retry requires a fresh
// live preflight.
func claimApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(
	receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt,
) (applyAllSourceRevisionAdvanceSelectedLivePreflightState, error) {
	state, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt)
	if err != nil {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, err
	}
	if !receipt.identity.claim.claimed.CompareAndSwap(false, true) {
		return applyAllSourceRevisionAdvanceSelectedLivePreflightState{}, fmt.Errorf(
			"selected live-preflight receipt was already claimed",
		)
	}
	return state, nil
}

func cloneApplyAllSourceRevisionAdvanceAttemptProvenance(
	provenance applyAllSourceRevisionAdvanceAttemptProvenance,
) applyAllSourceRevisionAdvanceAttemptProvenance {
	out := provenance
	out.admissionProvenance = cloneApplyAllSourceRevisionAdvanceExecutionProofProvenance(
		provenance.admissionProvenance,
	)
	return out
}

func cloneApplyAllSourceRevisionAdvanceSelectedLivePreflightProvenance(
	provenance applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance,
) applyAllSourceRevisionAdvanceSelectedLivePreflightProvenance {
	out := provenance
	out.preparedWitness = cloneApplyAllSourceRevisionAdvanceAttemptProvenance(
		provenance.preparedWitness,
	)
	return out
}