package controlplane

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

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

// applyAllSourceRevisionAdvanceSelectedAtomClaim is minted by Begin and shared
// by every later carrier for that exact installation incarnation. The selected
// atom execution leaf will consume it once.
type applyAllSourceRevisionAdvanceSelectedAtomClaim struct {
	claimed atomic.Bool
}

// applyAllSourceRevisionAdvanceAttempt is an opaque witness for one exact
// installation incarnation. A terminal transition must consume the complete
// admitted tuple, its selected proof entry, execution runtime, apply options,
// and installing row together.
type applyAllSourceRevisionAdvanceAttempt struct {
	identity *applyAllSourceRevisionAdvanceAttemptIdentity
	witness  applyAllSourceRevisionAdvanceAttemptWitness
}

type applyAllSourceRevisionAdvanceAttemptWitness struct {
	selectedAtomClaim *applyAllSourceRevisionAdvanceSelectedAtomClaim
}

type applyAllSourceRevisionAdvanceAttemptIdentity struct {
	admission          applyAllSourceRevisionAdvanceExecutionAdmission
	selectedProofIndex int
	pendingOrdinal     int
	installingRow      installedPackRow
	applyOptions       packs.ApplyOptions
	selectedAtomClaim  *applyAllSourceRevisionAdvanceSelectedAtomClaim
	provenance         applyAllSourceRevisionAdvanceAttemptProvenance
	minted             bool
}

// applyAllSourceRevisionAdvanceAttemptProvenance is detached from the retained
// admission. It detects coordinated edits that would otherwise leave that
// admission internally valid while changing the exact tuple or runtime that
// Begin accepted.
type applyAllSourceRevisionAdvanceAttemptProvenance struct {
	admissionIdentity          *applyAllSourceRevisionAdvanceExecutionAdmissionIdentity
	admissionProvenance        applyAllSourceRevisionAdvanceExecutionProofProvenance
	executionRuntime           *applyAllSourceRevisionAdvanceExecutionRuntime
	executionRuntimeProvenance *applyAllSourceRevisionAdvanceExecutionRuntimeProvenance
	executionRuntimeWitness    applyAllSourceRevisionAdvanceExecutionRuntimeProvenance
	selectedProofIndex         int
	pendingOrdinal             int
	installingRow              installedPackRow
	applyOptions               packs.ApplyOptions
	registrations              packs.ComponentKindRegistrationSet
	selectedAtomClaim          *applyAllSourceRevisionAdvanceSelectedAtomClaim
	minted                     bool
}

// prepareApplyAllSourceRevisionAdvanceAttempt detaches the complete admitted
// tuple and freezes the exact selection and execution inputs before Begin
// performs any schema read or mutation. The returned identity stays unminted
// until the fresh installing row has been constructed and validated.
func prepareApplyAllSourceRevisionAdvanceAttempt(
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
	pendingOrdinal int,
	applyOptions packs.ApplyOptions,
) (*applyAllSourceRevisionAdvanceAttemptIdentity, error) {
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		return nil, err
	}
	retainedAdmission := cloneApplyAllSourceRevisionAdvanceExecutionAdmission(admission)
	proof := retainedAdmission.identity.proof
	pendingIndexes := make([]int, 0, len(proof.entries))
	for index, entry := range proof.entries {
		if entry.current.pending != nil {
			pendingIndexes = append(pendingIndexes, index)
		}
	}
	if pendingOrdinal < 0 || pendingOrdinal >= len(pendingIndexes) {
		return nil, applyAllSourceRevisionAdvancePlanError(
			ApplyAllSourceRevisionAdvancePlanState, -1, "",
			fmt.Errorf("pending admission ordinal %d is out of range", pendingOrdinal),
		)
	}
	selectedProofIndex := pendingIndexes[pendingOrdinal]
	advance := proof.entries[selectedProofIndex].current.advance
	runtime := proof.executionRuntime
	if runtime == nil || runtime.provenance == nil {
		return nil, applyAllSourceRevisionAdvanceAttemptError(
			"prepare",
			advance, fmt.Errorf("admitted execution runtime provenance is unavailable"),
		)
	}
	identity := &applyAllSourceRevisionAdvanceAttemptIdentity{
		admission:          retainedAdmission,
		selectedProofIndex: selectedProofIndex,
		pendingOrdinal:     pendingOrdinal,
		applyOptions:       applyOptions,
		provenance: applyAllSourceRevisionAdvanceAttemptProvenance{
			admissionIdentity:          retainedAdmission.identity,
			admissionProvenance:        cloneApplyAllSourceRevisionAdvanceExecutionProofProvenance(retainedAdmission.identity.provenance),
			executionRuntime:           runtime,
			executionRuntimeProvenance: runtime.provenance,
			executionRuntimeWitness:    *runtime.provenance,
			selectedProofIndex:         selectedProofIndex,
			pendingOrdinal:             pendingOrdinal,
			applyOptions:               applyOptions,
			registrations:              runtime.componentKindRegistrationSet,
		},
	}
	if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(identity, false); err != nil {
		return nil, applyAllSourceRevisionAdvanceAttemptError("prepare", advance, err)
	}
	return identity, nil
}

// beginApplyAllSourceRevisionAdvanceAttempt consumes one exact selected-target
// live-preflight receipt and moves its admitted advance or retry to a fresh
// installing incarnation.
func beginApplyAllSourceRevisionAdvanceAttempt(
	ctx context.Context,
	receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt,
) (applyAllSourceRevisionAdvanceAttempt, error) {
	preflight, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt)
	if err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginReceiptError(err)
	}
	identityValue := *preflight.prepared
	identityValue.provenance = cloneApplyAllSourceRevisionAdvanceAttemptProvenance(
		preflight.prepared.provenance,
	)
	identity := &identityValue
	retained := identity.admission.identity
	db := preflight.database
	admitted := retained.proof.entries[identity.selectedProofIndex]
	entry := admitted.current
	advance := entry.advance
	if retained.provenance.activeRuntimeDatabase != db ||
		admitted.installedCompatibility != preflight.target ||
		identity.applyOptions != preflight.applyOptions {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("selected live-preflight authority changed before Begin"),
		)
	}
	if entry.pending == nil ||
		(entry.pending.Kind != ApplyAllSourceRevisionAdvancePendingAdvance &&
			entry.pending.Kind != ApplyAllSourceRevisionAdvancePendingRetry) {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("admitted pending kind is not advance or retry"),
		)
	}
	if strings.TrimSpace(entry.row.ID) == "" || strings.TrimSpace(entry.row.PackInstallID) == "" {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("installed row and prior pack-install incarnation identity are required"),
		)
	}
	if entry.row.Name != advance.Name {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("installed row is not the planned pack"),
		)
	}
	pending, err := classifyApplyAllSourceRevisionAdvance(
		entry.row, advance.SourceRevisionAdvanceAuthorization,
	)
	if err != nil || !equalApplyAllSourceRevisionAdvancePending(pending, entry.pending) {
		if err == nil {
			err = fmt.Errorf("pending classification does not match the frozen row")
		}
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(advance, err)
	}
	if err := validateApplyAllSourceRevisionAdvanceFrozenCurrentTargetEntry(entry, admitted.currentTarget); err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("validate admitted current target: %w", err),
		)
	}
	if err := validateApplyAllSourceRevisionAdvanceTarget(advance, admitted.installedCompatibility); err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("validate admitted installed target: %w", err),
		)
	}
	if err := packs.ValidateSourceBackedExecutionAtomTarget(admitted.atom, admitted.installedCompatibility); err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("validate admitted execution atom: %w", err),
		)
	}
	facts, err := admitted.atom.Facts()
	if err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("read admitted execution facts: %w", err),
		)
	}
	if facts.Name != entry.row.Name || string(facts.Category) != entry.row.Category {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("target changes installed pack name or category"),
		)
	}
	if facts.Category != packs.AppPack && facts.Category != packs.ConnectorPack {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("target category %q has no exact queue-ownership rule", facts.Category),
		)
	}
	runtimeDecl, err := packs.BuildPersistedRuntimeDeclaration(admitted.installedCompatibility)
	if err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("build target runtime declaration: %w", err),
		)
	}
	newPackInstallID := utils.GenerateID()
	if strings.TrimSpace(newPackInstallID) == "" || newPackInstallID == entry.row.PackInstallID {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("failed to mint a fresh pack-install incarnation identity"),
		)
	}
	meta := facts.Metadata
	installingRow := entry.row
	installingRow.PackInstallID = newPackInstallID
	installingRow.Category = string(facts.Category)
	installingRow.Status = installedPackStatusInstalling
	installingRow.IsDeleted = 0
	installingRow.PackVersion = sql.NullString{Valid: true, String: meta.PackVersion}
	installingRow.ManifestFormatVersion = sql.NullString{Valid: true, String: meta.ManifestFormatVersion}
	installingRow.PlatformCompatibility = nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid)
	installingRow.DesiredPayloadHash = sql.NullString{Valid: true, String: facts.PayloadHash}
	installingRow.PayloadHashAlgorithm = sql.NullString{Valid: true, String: meta.PayloadHashAlgorithm}
	installingRow.SourceKind = nullString(meta.SourceKind, meta.SourceKindValid)
	installingRow.SourceURI = nullString(meta.SourceURI, meta.SourceURIValid)
	installingRow.SourceRevision = nullString(meta.SourceRevision, meta.SourceRevisionValid)
	installingRow.HostRequirementsDigest = sql.NullString{Valid: true, String: facts.HostRequirementsDigest}
	if facts.Category == packs.ConnectorPack {
		installingRow.AppliedPayloadHash = sql.NullString{}
	}
	identity.installingRow = installingRow
	identity.provenance.installingRow = installingRow
	identity.selectedAtomClaim = &applyAllSourceRevisionAdvanceSelectedAtomClaim{}
	identity.provenance.selectedAtomClaim = identity.selectedAtomClaim
	identity.minted = true
	identity.provenance.minted = true
	if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(identity, true); err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(advance, err)
	}

	now := time.Now().UTC()
	assignments := []string{
		"PackInstallId = ?",
		"Category = ?",
		"Status = ?",
		"PackVersion = ?",
		"ManifestFormatVersion = ?",
		"PlatformCompatibility = ?",
		"DesiredPayloadHash = ?",
		"PayloadHashAlgorithm = ?",
		"SourceKind = ?",
		"SourceURI = ?",
		"SourceRevision = ?",
	}
	args := []any{
		newPackInstallID,
		string(facts.Category),
		installedPackStatusInstalling,
		meta.PackVersion,
		meta.ManifestFormatVersion,
		nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		facts.PayloadHash,
		meta.PayloadHashAlgorithm,
		nullString(meta.SourceKind, meta.SourceKindValid),
		nullString(meta.SourceURI, meta.SourceURIValid),
		nullString(meta.SourceRevision, meta.SourceRevisionValid),
	}
	if facts.Category == packs.ConnectorPack {
		assignments = append(assignments, "AppliedPayloadHash = NULL", "LastAppliedDate = NULL")
	}
	assignments = append(assignments,
		"HostRequirementsDigest = ?",
		"RuntimeObjectSpecDeclarations = ?",
		"RuntimeRelatedListRules = ?",
		"RuntimeSearchScopeOrder = ?",
		"RuntimeSearchAllDefaultFields = ?",
		"RuntimeFormulaFunctionBundles = ?",
		"RuntimeStartupShallowDeclarations = ?",
	)
	args = append(args,
		facts.HostRequirementsDigest,
		runtimeDecl.ObjectPayloadsJSON(),
		runtimeDecl.RelatedListRulesJSON(),
		runtimeDecl.SearchScopeOrderJSON(),
		runtimeDecl.SearchAllDefaultFieldsJSON(),
		runtimeDecl.FormulaFunctionBundlesJSON(),
		runtimeDecl.StartupShallowDeclaredRowsJSON(),
	)
	assignments = append(assignments,
		"InstallSource = 'apply'",
		"LastError = NULL",
		"LastModifiedDate = ?",
		"IsDeleted = 0",
	)
	args = append(args, now)

	predicate, predicateArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(
		entry.row, true,
	)
	args = append(args, predicateArgs...)
	claimed, err := claimApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt)
	if err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginReceiptError(err)
	}
	if claimed.prepared != preflight.prepared || claimed.database != preflight.database ||
		claimed.target != preflight.target || claimed.applyOptions != preflight.applyOptions ||
		!claimed.registrations.SameSnapshot(preflight.registrations) {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, fmt.Errorf("selected live-preflight authority changed before its one-shot claim"),
		)
	}
	err = withApplyAllSourceRevisionAdvanceExactIncarnationStore(
		ctx, db, entry.row, facts.Category,
		func(store applyAllSourceRevisionAdvanceExactIncarnationStore) error {
			if facts.Category == packs.ConnectorPack {
				if err := requireConnectorProjectionCheckpointBeforeRevision(ctx, store.tx, entry.row.PackInstallID); err != nil {
					return err
				}
			}
			update := `UPDATE _System_InstalledPack p
   SET ` + strings.Join(assignments, ",\n       ") + `
	WHERE ` + predicate + store.queueGuard()
			result, err := store.ExecContext(ctx, update, args...)
			if err != nil {
				return fmt.Errorf("write fresh installing incarnation: %w", err)
			}
			affected, err := result.RowsAffected()
			if err != nil {
				return fmt.Errorf("fresh installing incarnation rows affected: %w", err)
			}
			if affected != 1 {
				return fmt.Errorf(
					"fresh installing incarnation affected %d rows; frozen row changed or prior queue is not drained",
					affected,
				)
			}
			return nil
		},
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceAttempt{}, applyAllSourceRevisionAdvanceBeginError(
			advance, err,
		)
	}

	return applyAllSourceRevisionAdvanceAttempt{
		identity: identity,
		witness: applyAllSourceRevisionAdvanceAttemptWitness{
			selectedAtomClaim: identity.selectedAtomClaim,
		},
	}, nil
}

func applyAllSourceRevisionAdvanceBeginReceiptError(err error) error {
	return applyAllSourceRevisionAdvancePlanError(
		ApplyAllSourceRevisionAdvancePlanState, -1, "",
		fmt.Errorf("begin exact installation attempt: %w", err),
	)
}

func applyAllSourceRevisionAdvanceBeginError(
	advance normalizedApplyAllSourceRevisionAdvance,
	err error,
) error {
	return applyAllSourceRevisionAdvancePlanError(
		ApplyAllSourceRevisionAdvancePlanState, advance.index, advance.Name,
		fmt.Errorf("begin exact installation attempt: %w", err),
	)
}