package controlplane

import (
	"context"
	"errors"
	"fmt"
	"time"

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

type applyAllSourceRevisionAdvanceSuccess struct {
	connectedAppChangedIDs []string
}

type applyAllSourceRevisionAdvanceFinalizationAuthority struct {
	identity *applyAllSourceRevisionAdvanceFinalizationAuthorityIdentity
	witness  applyAllSourceRevisionAdvanceFinalizationAuthorityProvenance
}

type applyAllSourceRevisionAdvanceFinalizationAuthorityIdentity struct {
	readiness       applyAllSourceRevisionAdvanceActivationReadiness
	activationClaim *applyAllSourceRevisionAdvanceActivationClaim
	provenance      *applyAllSourceRevisionAdvanceFinalizationAuthorityProvenance
	minted          bool
}

type applyAllSourceRevisionAdvanceFinalizationAuthorityProvenance struct {
	readinessIdentity *applyAllSourceRevisionAdvanceActivationReadinessIdentity
	activationClaim   *applyAllSourceRevisionAdvanceActivationClaim
}

type applyAllSourceRevisionAdvanceFinalizationAuthorityState struct {
	readiness applyAllSourceRevisionAdvanceActivationReadinessState
	attempt   applyAllSourceRevisionAdvanceAttempt
}

type applyAllSourceRevisionAdvanceFinalizationOperations struct {
	finalize func(
		context.Context,
		applyAllSourceRevisionAdvanceFinalizationAuthority,
	) (applyAllSourceRevisionAdvanceSuccess, error)
	compensate func(
		context.Context,
		applyAllSourceRevisionAdvanceRuntimeRegistration,
		error,
	) error
}

// completeApplyAllSourceRevisionAdvanceExecution consumes one post-maintenance
// readiness carrier exactly once, activates only the fresh incarnation minted
// by Begin, and compensates any determinate terminal-write failure. A mere
// successful atom execution cannot reach this boundary. The ApplyAll
// orchestrator owns post-commit notification and ambiguous reconciliation.
func completeApplyAllSourceRevisionAdvanceExecution(
	ctx context.Context,
	readiness applyAllSourceRevisionAdvanceActivationReadiness,
) (applyAllSourceRevisionAdvanceSuccess, error) {
	return completeApplyAllSourceRevisionAdvanceExecutionWith(
		ctx,
		readiness,
		applyAllSourceRevisionAdvanceFinalizationOperations{
			finalize:   commitApplyAllSourceRevisionAdvanceExecutionActivation,
			compensate: compensateApplyAllSourceRevisionAdvanceExecution,
		},
	)
}

func completeApplyAllSourceRevisionAdvanceExecutionWith(
	ctx context.Context,
	readiness applyAllSourceRevisionAdvanceActivationReadiness,
	operations applyAllSourceRevisionAdvanceFinalizationOperations,
) (applyAllSourceRevisionAdvanceSuccess, error) {
	state, err := requireApplyAllSourceRevisionAdvanceActivationReadiness(readiness)
	if err != nil {
		return applyAllSourceRevisionAdvanceSuccess{}, err
	}
	registration := readiness.identity.execution.identity.registration
	advance := state.execution.registration.snapshots.attempt.advance
	if operations.finalize == nil || operations.compensate == nil {
		return applyAllSourceRevisionAdvanceSuccess{}, applyAllSourceRevisionAdvanceAttemptError(
			"complete", advance, fmt.Errorf("finalization operations are incomplete"),
		)
	}
	if !state.activationClaim.claimed.CompareAndSwap(false, true) {
		return applyAllSourceRevisionAdvanceSuccess{}, applyAllSourceRevisionAdvanceAttemptError(
			"complete", advance, fmt.Errorf("successful execution was already finalized"),
		)
	}

	authority, err := mintApplyAllSourceRevisionAdvanceFinalizationAuthority(readiness)
	if err != nil {
		return applyAllSourceRevisionAdvanceSuccess{}, compensateApplyAllSourceRevisionAdvanceOperationError(
			ctx, registration, advance, "complete", err, operations.compensate,
		)
	}
	success, err := operations.finalize(ctx, authority)
	if err != nil {
		var commitErr *applyAllSourceRevisionAdvanceExactIncarnationCommitError
		if errors.As(err, &commitErr) {
			// Commit failure is ambiguous: activation and every transactional
			// side effect may already be durable. Keep the candidate runtime in
			// place and let a later lock-held whole-tuple reread classify the row.
			// Compensating here could roll runtime metadata back while the exact
			// installation row is already active.
			ambiguousErr, mintErr := mintApplyAllSourceRevisionAdvanceAmbiguousActivationError(
				authority, commitErr, success,
			)
			if mintErr != nil {
				return applyAllSourceRevisionAdvanceSuccess{}, applyAllSourceRevisionAdvanceAttemptError(
					"complete", advance, errors.Join(err, fmt.Errorf(
						"mint ambiguous activation reconciliation authority: %w", mintErr,
					)),
				)
			}
			return applyAllSourceRevisionAdvanceSuccess{}, applyAllSourceRevisionAdvanceAttemptError(
				"complete", advance, ambiguousErr,
			)
		}
		return applyAllSourceRevisionAdvanceSuccess{}, compensateApplyAllSourceRevisionAdvanceOperationError(
			ctx, registration, advance, "complete", err, operations.compensate,
		)
	}
	return success, nil
}

func mintApplyAllSourceRevisionAdvanceFinalizationAuthority(
	readiness applyAllSourceRevisionAdvanceActivationReadiness,
) (applyAllSourceRevisionAdvanceFinalizationAuthority, error) {
	state, err := requireApplyAllSourceRevisionAdvanceActivationReadiness(readiness)
	if err != nil {
		return applyAllSourceRevisionAdvanceFinalizationAuthority{}, err
	}
	if !state.activationClaim.claimed.Load() {
		return applyAllSourceRevisionAdvanceFinalizationAuthority{}, fmt.Errorf(
			"post-maintenance readiness activation claim was not consumed",
		)
	}
	identity := &applyAllSourceRevisionAdvanceFinalizationAuthorityIdentity{
		readiness:       readiness,
		activationClaim: state.activationClaim,
		minted:          true,
	}
	identity.provenance = &applyAllSourceRevisionAdvanceFinalizationAuthorityProvenance{
		readinessIdentity: readiness.identity,
		activationClaim:   state.activationClaim,
	}
	authority := applyAllSourceRevisionAdvanceFinalizationAuthority{
		identity: identity,
		witness:  *identity.provenance,
	}
	if _, err := requireApplyAllSourceRevisionAdvanceFinalizationAuthority(authority); err != nil {
		return applyAllSourceRevisionAdvanceFinalizationAuthority{}, err
	}
	return authority, nil
}

func requireApplyAllSourceRevisionAdvanceFinalizationAuthority(
	authority applyAllSourceRevisionAdvanceFinalizationAuthority,
) (applyAllSourceRevisionAdvanceFinalizationAuthorityState, error) {
	identity := authority.identity
	if identity == nil || !identity.minted || identity.provenance == nil {
		return applyAllSourceRevisionAdvanceFinalizationAuthorityState{}, fmt.Errorf(
			"finalization authority was not minted from a successful execution",
		)
	}
	if identity.readiness.identity == nil ||
		identity.readiness.identity != identity.provenance.readinessIdentity ||
		identity.readiness.identity != authority.witness.readinessIdentity {
		return applyAllSourceRevisionAdvanceFinalizationAuthorityState{}, fmt.Errorf(
			"finalization authority changed activation-readiness identity",
		)
	}
	if identity.activationClaim == nil ||
		identity.activationClaim != identity.readiness.identity.activationClaim ||
		identity.activationClaim != identity.provenance.activationClaim ||
		identity.activationClaim != authority.witness.activationClaim ||
		!identity.activationClaim.claimed.Load() {
		return applyAllSourceRevisionAdvanceFinalizationAuthorityState{}, fmt.Errorf(
			"finalization authority changed or did not consume its activation claim",
		)
	}
	readinessState, err := requireApplyAllSourceRevisionAdvanceActivationReadiness(identity.readiness)
	if err != nil {
		return applyAllSourceRevisionAdvanceFinalizationAuthorityState{}, fmt.Errorf(
			"validate finalization activation readiness: %w", err,
		)
	}
	registration := identity.readiness.identity.execution.identity.registration
	return applyAllSourceRevisionAdvanceFinalizationAuthorityState{
		readiness: readinessState,
		attempt:   registration.identity.snapshots.identity.attempt,
	}, nil
}

// commitApplyAllSourceRevisionAdvanceExecutionActivation performs the exact
// terminal transaction and accepts only post-execution finalization authority.
func commitApplyAllSourceRevisionAdvanceExecutionActivation(
	ctx context.Context,
	authority applyAllSourceRevisionAdvanceFinalizationAuthority,
) (applyAllSourceRevisionAdvanceSuccess, error) {
	authorityState, err := requireApplyAllSourceRevisionAdvanceFinalizationAuthority(authority)
	if err != nil {
		return applyAllSourceRevisionAdvanceSuccess{}, err
	}
	state, err := requireApplyAllSourceRevisionAdvanceAttempt(authorityState.attempt, "complete")
	if err != nil {
		return applyAllSourceRevisionAdvanceSuccess{}, err
	}

	row := state.identity.installingRow
	category := state.facts.Category
	runtimeDecl, err := packs.BuildPersistedRuntimeDeclaration(state.target)
	if err != nil {
		return applyAllSourceRevisionAdvanceSuccess{}, applyAllSourceRevisionAdvanceAttemptError(
			"complete", state.advance, fmt.Errorf("build exact target runtime declaration: %w", err),
		)
	}
	predicate, predicateArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
	for _, runtime := range []struct {
		column string
		value  string
	}{
		{column: "p.RuntimeObjectSpecDeclarations", value: runtimeDecl.ObjectPayloadsJSON()},
		{column: "p.RuntimeRelatedListRules", value: runtimeDecl.RelatedListRulesJSON()},
		{column: "p.RuntimeSearchScopeOrder", value: runtimeDecl.SearchScopeOrderJSON()},
		{column: "p.RuntimeSearchAllDefaultFields", value: runtimeDecl.SearchAllDefaultFieldsJSON()},
		{column: "p.RuntimeFormulaFunctionBundles", value: runtimeDecl.FormulaFunctionBundlesJSON()},
		{column: "p.RuntimeStartupShallowDeclarations", value: runtimeDecl.StartupShallowDeclaredRowsJSON()},
	} {
		predicate += "\n   AND " + applyAllSourceRevisionAdvanceExactJSONPredicate(runtime.column)
		predicateArgs = append(predicateArgs, runtime.value)
	}
	now := time.Now().UTC()
	args := []any{installedPackStatusActive, state.facts.PayloadHash, now, now}
	args = append(args, predicateArgs...)

	var connectedAppChangedIDs []string
	err = withApplyAllSourceRevisionAdvanceExactIncarnationStore(
		ctx, state.database, row, category,
		func(store applyAllSourceRevisionAdvanceExactIncarnationStore) error {
			update := `UPDATE _System_InstalledPack p
   SET Status = ?,
       AppliedPayloadHash = ?,
       LastAppliedDate = ?,
       LastError = NULL,
       LastModifiedDate = ?,
       IsDeleted = 0
 WHERE ` + predicate + store.queueGuard()
			result, err := store.ExecContext(ctx, update, args...)
			if err != nil {
				return fmt.Errorf("activate exact installation incarnation: %w", err)
			}
			affected, err := result.RowsAffected()
			if err != nil {
				return fmt.Errorf("active installation incarnation rows affected: %w", err)
			}
			if affected != 1 {
				return fmt.Errorf(
					"active installation incarnation affected %d rows; exact installing row changed",
					affected,
				)
			}
			if err := commitSourceRevisionAdvancePackGeneration(
				ctx,
				store.tx,
				state.advance,
				appliedAppPack{
					pack:        state.target,
					payloadHash: state.facts.PayloadHash,
				},
			); err != nil {
				return fmt.Errorf("publish source-backed pack generation: %w", err)
			}
			if err := upsertPackDependencyRows(ctx, store, state.target, now); err != nil {
				return fmt.Errorf("reconcile active pack dependencies: %w", err)
			}
			if err := refreshConnectorInboundProjectionSnapshotsAfterPackActivated(
				ctx, store, state.target, now,
			); err != nil {
				return fmt.Errorf("refresh active connector inbound projections: %w", err)
			}
			if category == packs.ConnectorPack {
				if err := ensureEmptyConnectorProjectionProtocolTx(ctx, store.tx, row.Name); err != nil {
					return fmt.Errorf("initialize active connector projection protocol: %w", err)
				}
			}
			connectedAppChangedIDs, err = store.publishSourceBackedConnectedAppsForPack(
				ctx, row.Name,
			)
			if err != nil {
				return fmt.Errorf("publish active connected apps: %w", err)
			}
			return nil
		},
	)
	if err != nil {
		var commitErr *applyAllSourceRevisionAdvanceExactIncarnationCommitError
		if errors.As(err, &commitErr) {
			return applyAllSourceRevisionAdvanceSuccess{
				connectedAppChangedIDs: append([]string(nil), connectedAppChangedIDs...),
			}, applyAllSourceRevisionAdvanceAttemptError("complete", state.advance, err)
		}
		return applyAllSourceRevisionAdvanceSuccess{},
			applyAllSourceRevisionAdvanceAttemptError("complete", state.advance, err)
	}
	return applyAllSourceRevisionAdvanceSuccess{
		connectedAppChangedIDs: append([]string(nil), connectedAppChangedIDs...),
	}, nil
}

func cloneApplyAllSourceRevisionAdvanceSuccess(
	success applyAllSourceRevisionAdvanceSuccess,
) applyAllSourceRevisionAdvanceSuccess {
	return applyAllSourceRevisionAdvanceSuccess{
		connectedAppChangedIDs: append([]string(nil), success.connectedAppChangedIDs...),
	}
}