package controlplane

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"

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

// applyAllSourceRevisionAdvances is the sole production owner of the exact
// source-revision execution chain. It performs at most one durable mutation per
// proof, then reconstructs and rereads the complete tuple before selecting the
// next operation. The caller-provided scope keeps every proof and any ambiguous
// terminal reconciliation inside the same ApplyAll advisory-lock lifetime.
func applyAllSourceRevisionAdvances(
	ctx context.Context,
	r *Runner,
	options InstallPackOptions,
	applyLock controlPlaneApplyLockScope,
) error {
	database := runnerSQLDB(r)
	if database == nil {
		return fmt.Errorf("source revision advance runner database is unavailable")
	}
	if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
		return fmt.Errorf("source revision advance requires the active ApplyAll lock: %w", err)
	}
	plan, err := normalizeApplyAllSourceRevisionAdvancePlan(
		options.ApplyAllSourceRevisionAdvances,
	)
	if err != nil {
		return err
	}
	if err := markOfflineCandidateCompleteSourceRevisionReplays(
		plan, options.offlineCandidateCompleteSourceRevisionReplays,
	); err != nil {
		return err
	}
	if len(plan) == 0 {
		return nil
	}
	applyOptions := packs.ApplyOptions{
		SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
		RepairLegacyUniqueDuplicates:         options.RepairLegacyUniqueDuplicates,
		DeferFormulaViewRebuild:              options.DeferFormulaViewRebuild,
	}

	for completedMutations := 0; ; completedMutations++ {
		if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
			return fmt.Errorf("source revision advance lost the ApplyAll lock: %w", err)
		}
		admission, err := composeApplyAllSourceRevisionAdvanceAdmission(ctx, r, plan)
		if err != nil {
			return err
		}
		if err := preflightApplyAllSourceRevisionAdvanceGenerationAuthorities(
			ctx, database, admission,
		); err != nil {
			return err
		}
		pendingCount := applyAllSourceRevisionAdvanceAdmissionPendingCount(admission)
		if pendingCount == 0 {
			return nil
		}
		if completedMutations >= len(plan) {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState, -1, "",
				fmt.Errorf(
					"source revision plan still has %d pending operation(s) after %d exact mutation(s)",
					pendingCount, completedMutations,
				),
			)
		}

		_, selected, err := applyAllSourceRevisionAdvanceAdmittedPendingEntry(admission, 0)
		if err != nil {
			return err
		}
		switch selected.current.pending.Kind {
		case ApplyAllSourceRevisionAdvancePendingCanonicalize:
			if err := canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
				ctx, admission, 0,
			); err != nil {
				return err
			}
		case ApplyAllSourceRevisionAdvancePendingAdvance,
			ApplyAllSourceRevisionAdvancePendingRetry:
			if err := executeApplyAllSourceRevisionAdvanceAdmission(
				ctx, r, admission, applyOptions, options.CandidateFinalVerify, options.offlineCandidate, applyLock,
			); err != nil {
				return err
			}
		default:
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState,
				selected.current.advance.index,
				selected.current.advance.Name,
				fmt.Errorf("unsupported admitted pending kind %q", selected.current.pending.Kind),
			)
		}
	}
}

// preflightApplyAllSourceRevisionAdvanceGenerationAuthorities classifies every
// admitted entry before the orchestrator trusts a no-op result or mutates one
// selected entry. A nil pending operation means only that the mutable source
// revision already matches; it is not proof that immutable authority is absent,
// exact, or internally consistent.
func preflightApplyAllSourceRevisionAdvanceGenerationAuthorities(
	ctx context.Context,
	database *sql.DB,
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
) error {
	if database == nil {
		return fmt.Errorf("source revision generation preflight database is unavailable")
	}
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		return err
	}
	for _, entry := range admission.identity.proof.entries {
		advance := entry.current.advance
		current := entry.replayBase
		if current == nil {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState,
				advance.index,
				advance.Name,
				fmt.Errorf("revision-bound replay base is unavailable"),
			)
		}
		target := entry.installedCompatibility
		requiresTransition := entry.current.pending != nil &&
			(entry.current.pending.Kind == ApplyAllSourceRevisionAdvancePendingAdvance ||
				entry.current.pending.Kind == ApplyAllSourceRevisionAdvancePendingRetry)
		err := verifySourceRevisionAdvanceGenerationAuthority(
			ctx,
			database,
			advance,
			current,
			target,
			entry.current.row,
			requiresTransition,
		)
		if err != nil {
			return applyAllSourceRevisionAdvancePlanError(
				ApplyAllSourceRevisionAdvancePlanState,
				advance.index,
				advance.Name,
				fmt.Errorf("preflight immutable generation authority: %w", err),
			)
		}
	}
	return nil
}

func composeApplyAllSourceRevisionAdvanceAdmission(
	ctx context.Context,
	r *Runner,
	plan []normalizedApplyAllSourceRevisionAdvance,
) (applyAllSourceRevisionAdvanceExecutionAdmission, error) {
	precondition, err := proveApplyAllSourceRevisionAdvancePreconditions(ctx, r, plan)
	if err != nil {
		return applyAllSourceRevisionAdvanceExecutionAdmission{}, err
	}
	sourceOptions, ok := packs.OperationPackSourceOptions(ctx)
	if !ok {
		sourceOptions = packs.DefaultPackSourceOptions()
	}
	source, err := resolveApplyAllSourceRevisionAdvanceSourceTupleWithBase(
		plan, r.componentKindRegistrationSet(), sourceOptions,
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceExecutionAdmission{}, err
	}
	compatibility, err := resolveApplyAllSourceRevisionAdvanceInstalledTuple(
		ctx, r, precondition, source,
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceExecutionAdmission{}, err
	}
	proof, err := bindApplyAllSourceRevisionAdvanceExecutionProof(
		r, precondition, source, compatibility,
	)
	if err != nil {
		return applyAllSourceRevisionAdvanceExecutionAdmission{}, err
	}
	return admitApplyAllSourceRevisionAdvanceExecution(ctx, runnerSQLDB(r), proof)
}

func applyAllSourceRevisionAdvanceAdmissionPendingCount(
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
) int {
	if admission.identity == nil {
		return 0
	}
	count := 0
	for _, entry := range admission.identity.proof.entries {
		if entry.current.pending != nil {
			count++
		}
	}
	return count
}

func executeApplyAllSourceRevisionAdvanceAdmission(
	ctx context.Context,
	r *Runner,
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
	applyOptions packs.ApplyOptions,
	candidateFinalVerify CandidateFinalVerifier,
	offlineCandidate *offlineCandidateExecution,
	applyLock controlPlaneApplyLockScope,
) error {
	prepared, err := prepareApplyAllSourceRevisionAdvanceAttempt(
		admission, 0, applyOptions,
	)
	if err != nil {
		return err
	}
	_, selected, err := applyAllSourceRevisionAdvanceAdmittedPendingEntry(admission, 0)
	if err != nil {
		return err
	}
	packName := strings.TrimSpace(selected.current.advance.Name)
	var (
		execution    applyAllSourceRevisionAdvanceExecution
		registration applyAllSourceRevisionAdvanceRuntimeRegistration
	)
	preflightAndExecuteSelected := func(executeCtx context.Context) error {
		receipt, preflightErr := preflightApplyAllSourceRevisionAdvanceSelectedTarget(
			executeCtx, prepared, offlineCandidate,
		)
		if preflightErr != nil {
			return preflightErr
		}
		attempt, beginErr := beginApplyAllSourceRevisionAdvanceAttempt(executeCtx, receipt)
		if beginErr != nil {
			return beginErr
		}
		snapshots, snapshotErr := captureApplyAllSourceRevisionAdvanceSnapshots(executeCtx, attempt)
		if snapshotErr != nil {
			if failErr := failApplyAllSourceRevisionAdvanceAttempt(executeCtx, attempt, snapshotErr); failErr != nil {
				return errors.Join(snapshotErr, failErr)
			}
			return snapshotErr
		}
		registration, err = registerApplyAllSourceRevisionAdvanceRuntime(executeCtx, snapshots)
		if err != nil {
			if failErr := failApplyAllSourceRevisionAdvanceAttempt(executeCtx, attempt, err); failErr != nil {
				return errors.Join(err, failErr)
			}
			return err
		}
		execution, err = executeApplyAllSourceRevisionAdvanceSelectedAtom(executeCtx, registration)
		return err
	}
	// The exact candidate scope begins only after the selected immutable
	// admission is known, covers every internal check that may authorize the
	// transition, and is revoked before the caller-provided final verifier.
	err = runOfflineCandidateFormulaStorageTransition(
		ctx, offlineCandidate, packName, preflightAndExecuteSelected,
	)
	if err != nil {
		return err
	}
	readiness, err := prepareApplyAllSourceRevisionAdvanceActivation(ctx, r, execution, candidateFinalVerify)
	if err != nil {
		return err
	}
	success, err := completeApplyAllSourceRevisionAdvanceExecution(ctx, readiness)
	if err != nil {
		var ambiguous *applyAllSourceRevisionAdvanceAmbiguousActivationError
		if !errors.As(err, &ambiguous) {
			return err
		}
		reconciliation, reconciliationErr :=
			reconcileApplyAllSourceRevisionAdvanceAmbiguousActivationUnderApplyLock(
				ctx, applyLock, runnerSQLDB(r), err,
			)
		if reconciliationErr != nil {
			return errors.Join(err, reconciliationErr)
		}
		switch reconciliation.outcome {
		case applyAllSourceRevisionAdvanceActivationCommitted:
			success = reconciliation.success
		case applyAllSourceRevisionAdvanceActivationNotCommitted:
			return compensateApplyAllSourceRevisionAdvanceExecution(ctx, registration, err)
		default:
			return errors.Join(err, fmt.Errorf(
				"ambiguous activation reconciliation returned unsupported outcome %q",
				reconciliation.outcome,
			))
		}
	}

	packs.NotifySourceBackedConnectedAppChanges(
		r.PackDepsWithOptions(applyOptions), success.connectedAppChangedIDs,
	)
	if err := reconcileConnectorInboundWorkersAfterPackLifecycle(ctx, r); err != nil {
		return fmt.Errorf("reconcile connector inbound workers after source revision advance: %w", err)
	}
	// The next admission/preflight is a dependent pack boundary. Rebuild the
	// complete Runner view from the now-active exact incarnation before proving
	// another plan entry, matching ApplyPacks' inter-app-pack refresh contract.
	if err := r.RefreshCache(ctx); err != nil {
		return fmt.Errorf("refresh cache after source revision advance: %w", err)
	}
	return nil
}