// Copyright 2025 Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package frontend

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"

	"k8s.io/apimachinery/pkg/api/operation"
	k8sutilruntime "k8s.io/apimachinery/pkg/util/runtime"
	utilsclock "k8s.io/utils/clock"

	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"

	arohcpv1alpha1 "github.com/openshift-online/ocm-sdk-go/arohcp/v1alpha1"
	cmv1 "github.com/openshift-online/ocm-sdk-go/clustersmgmt/v1"

	"github.com/Azure/ARO-HCP/frontend/pkg/metrics"
	"github.com/Azure/ARO-HCP/internal/admission"
	"github.com/Azure/ARO-HCP/internal/api/coreapi"
	"github.com/Azure/ARO-HCP/internal/api/metadataapi"
	"github.com/Azure/ARO-HCP/internal/audit"
	"github.com/Azure/ARO-HCP/internal/azureapi/v20240610preview"
	"github.com/Azure/ARO-HCP/internal/azureapi/v20251223preview"
	"github.com/Azure/ARO-HCP/internal/azureapi/v20260630preview"
	"github.com/Azure/ARO-HCP/internal/azureapi/v20260901preview"
	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/corecosmosstorage"
	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils"
	"github.com/Azure/ARO-HCP/internal/ocm"
	"github.com/Azure/ARO-HCP/internal/utils"
	"github.com/Azure/ARO-HCP/internal/utils/armhelpers"
	"github.com/Azure/ARO-HCP/internal/validation"
)

type Frontend struct {
	clock                utilsclock.PassiveClock
	clusterServiceClient ocm.ClusterServiceClientSpec
	listener             net.Listener
	metricsListener      net.Listener
	server               http.Server
	metricsServer        http.Server
	resourcesDBClient    corecosmosstorage.ResourcesDBClient
	auditClient          audit.Client
	collector            *metrics.SubscriptionCollector
	healthGauge          prometheus.Gauge
	// this is the azure location for this instance of the frontend
	azureLocation string

	apiRegistry coreapi.APIRegistry

	exitOnPanic bool
}

func NewFrontend(
	logger logr.Logger,
	listener net.Listener,
	metricsListener net.Listener,
	registerer prometheus.Registerer,
	gatherer prometheus.Gatherer,
	resourcesDBClient corecosmosstorage.ResourcesDBClient,
	csClient ocm.ClusterServiceClientSpec,
	auditClient audit.Client,
	azureLocation string,
	exitOnPanic bool,
) *Frontend {
	// zero side-effect registration path
	apiRegistry := coreapi.NewAPIRegistry()
	metadataapi.Must[any](nil, v20240610preview.RegisterVersion(apiRegistry))
	metadataapi.Must[any](nil, v20251223preview.RegisterVersion(apiRegistry))
	metadataapi.Must[any](nil, v20260630preview.RegisterVersion(apiRegistry))
	metadataapi.Must[any](nil, v20260901preview.RegisterVersion(apiRegistry))

	f := &Frontend{
		clock:                utilsclock.RealClock{},
		clusterServiceClient: csClient,
		listener:             listener,
		metricsListener:      metricsListener,
		server: http.Server{
			BaseContext: func(net.Listener) context.Context {
				ctx := context.Background()
				ctx = utils.ContextWithLogger(ctx, logger)
				return ctx
			},
		},
		metricsServer: http.Server{
			BaseContext: func(net.Listener) context.Context {
				return utils.ContextWithLogger(context.Background(), logger)
			},
		},
		auditClient:       auditClient,
		resourcesDBClient: resourcesDBClient,
		collector:         metrics.NewSubscriptionCollector(registerer, resourcesDBClient, azureLocation),
		healthGauge: promauto.With(registerer).NewGauge(
			prometheus.GaugeOpts{
				Name: healthGaugeName,
				Help: "Reports the health status of the service (0: not healthy, 1: healthy).",
			},
		),
		azureLocation: azureLocation,
		apiRegistry:   apiRegistry,
		exitOnPanic:   exitOnPanic,
	}

	f.server.Handler = f.routes(registerer)
	f.metricsServer.Handler = f.metricsRoutes(gatherer)

	return f
}

func (f *Frontend) Run(ctx context.Context) error {
	ctx, cancel := context.WithCancelCause(ctx)
	defer func() {
		cancel(fmt.Errorf("run returned"))

		// always attempt a graceful shutdown, a double ctrl+c exits the process
		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 31*time.Second)
		defer shutdownCancel()
		_ = f.server.Shutdown(shutdownCtx)
		_ = f.metricsServer.Shutdown(shutdownCtx)
	}()

	if len(f.azureLocation) == 0 {
		panic("azureLocation must be set")
	}

	// We set k8s.io/apimachinery/pkg/util/runtime.ReallyCrash to the value of the ExitOnPanic option to
	// control the behavior of k8s.io/apimachinery/pkg/util/runtime.HandleCrash* methods
	k8sutilruntime.ReallyCrash = f.exitOnPanic

	// This just digs up the logger passed to NewFrontend.
	logger := utils.LoggerFromContext(ctx)

	logger.Info(fmt.Sprintf("listening on %s", f.listener.Addr().String()))
	logger.Info(fmt.Sprintf("metrics listening on %s", f.metricsListener.Addr().String()))

	errCh := make(chan error, 2)
	wg := sync.WaitGroup{}
	wg.Add(3)
	go func() {
		defer k8sutilruntime.HandleCrash()
		defer wg.Done()
		errCh <- f.server.Serve(f.listener)
	}()
	go func() {
		defer k8sutilruntime.HandleCrash()
		defer wg.Done()
		errCh <- f.metricsServer.Serve(f.metricsListener)
	}()
	go func() {
		defer k8sutilruntime.HandleCrash()
		defer wg.Done()
		f.collector.Run(ctx)
	}()

	<-ctx.Done()

	// always attempt a graceful shutdown, a double ctrl+c exits the process
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 31*time.Second)
	defer shutdownCancel()
	if err := f.server.Shutdown(shutdownCtx); err != nil {
		logger.Error(err, "failed to shutdown http server")
	}
	if err := f.metricsServer.Shutdown(shutdownCtx); err != nil {
		logger.Error(err, "failed to shutdown http server")
	}

	wg.Wait()
	close(errCh)
	errs := []error{}
	for err := range errCh {
		if err != nil {
			logger.Info("go func completed", "message", err.Error())
		}
		if !errors.Is(err, http.ErrServerClosed) {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

func (f *Frontend) NotFound(writer http.ResponseWriter, request *http.Request) {
	coreapi.WriteError(
		writer, http.StatusNotFound,
		coreapi.CloudErrorCodeNotFound, "",
		"The requested path could not be found.")
}

func (f *Frontend) Healthz(writer http.ResponseWriter, request *http.Request) {
	writer.WriteHeader(http.StatusOK)
	f.healthGauge.Set(1.0)
}

func (f *Frontend) Location(writer http.ResponseWriter, request *http.Request) {
	// This is strictly for development environments to help discover
	// the frontend's Azure region when port forwarding with kubectl.
	// e.g. LOCATION=$(curl http://localhost:8443/location)
	_, _ = writer.Write([]byte(f.azureLocation))
}

func dbListOptionsFromRequest(request *http.Request) *cosmosstorageutils.DBClientListResourceDocsOptions {
	// FIXME We may want to cap pageSizeHint. If we get a large enough
	//       $top argument (and there's enough actual clusters to reach
	//       that), we could potentially hit the 8MB response size limit.

	options := &cosmosstorageutils.DBClientListResourceDocsOptions{
		PageSizeHint: metadataapi.Ptr(int32(20)),
	}

	// The Resource Provider Contract implies $top is only honored when
	// following a "nextLink" after the initial collection GET request.
	// So only check for it when the URL includes a $skipToken.
	urlQuery := request.URL.Query()
	if urlQuery.Has("$skipToken") {
		options.ContinuationToken = metadataapi.Ptr(urlQuery.Get("$skipToken"))
		top, err := strconv.ParseInt(urlQuery.Get("$top"), 10, 32)
		if err == nil && top > 0 {
			options.PageSizeHint = metadataapi.Ptr(int32(top))
		}
	}
	return options
}

func (f *Frontend) ArmResourceListVersion(writer http.ResponseWriter, request *http.Request) error {
	ctx := request.Context()

	versionedInterface, err := VersionFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	subscriptionID := request.PathValue(PathSegmentSubscriptionID)
	location := request.PathValue(PathSegmentLocation)

	pagedResponse := coreapi.NewPagedResponse()

	csIterator := f.clusterServiceClient.ListVersions()
	for csVersion := range csIterator.Items(ctx) {
		versionName := strings.Replace(csVersion.ID(), metadataapi.OpenShiftVersionPrefix, "", 1)
		stringResource := "/subscriptions/" + subscriptionID + "/providers/" + coreapi.ProviderNamespace +
			"/locations/" + location + "/" + coreapi.VersionResourceTypeName + "/" + versionName
		resourceID, err := azcorearm.ParseResourceID(stringResource)
		if err != nil {
			return utils.TrackError(err)
		}
		value, err := marshalCSVersion(resourceID, csVersion, versionedInterface)
		if err != nil {
			return utils.TrackError(err)
		}
		pagedResponse.AddValue(value)
	}
	err = csIterator.GetError()

	// Check for iteration error.
	if err != nil {
		return utils.TrackError(err)
	}

	_, err = coreapi.WriteJSONResponse(writer, http.StatusOK, pagedResponse)
	if err != nil {
		return utils.TrackError(err)
	}
	return nil
}

// GetOpenshiftVersions implements the GET single resource API contract for ARM
// * 200 If the resource exists
// * 404 If the resource does not exist
func (f *Frontend) GetOpenshiftVersions(writer http.ResponseWriter, request *http.Request) error {
	ctx := request.Context()

	versionedInterface, err := VersionFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}
	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	versionName := resourceID.Name
	version, err := f.clusterServiceClient.GetVersion(ctx, versionName)
	if err != nil {
		return utils.TrackError(err)
	}
	responseBody, err := marshalCSVersion(resourceID, version, versionedInterface)
	if err != nil {
		return utils.TrackError(err)
	}

	_, err = coreapi.WriteJSONResponse(writer, http.StatusOK, responseBody)
	if err != nil {
		return utils.TrackError(err)
	}
	return nil
}

func (f *Frontend) ArmResourceActionRequestAdminCredential(writer http.ResponseWriter, request *http.Request) error {
	const operationRequest = cosmosstorageutils.OperationRequestSystemAdminCredentialRequest

	ctx := request.Context()

	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	// Parent resource is the hcpOpenShiftCluster.
	clusterResourceID := resourceID.Parent

	correlationData, err := CorrelationDataFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	cluster, err := f.resourcesDBClient.HCPClusters(clusterResourceID.SubscriptionID, clusterResourceID.ResourceGroupName).Get(ctx, clusterResourceID.Name)
	if err != nil {
		return utils.TrackError(err)
	}
	// CheckForProvisioningStateConflict does not log conflict errors
	// but does log unexpected errors like database failures.
	if err := checkForProvisioningStateConflict(ctx, f.resourcesDBClient, operationRequest, cluster.ID, cluster.ServiceProviderProperties.ProvisioningState); err != nil {
		return utils.TrackError(err)
	}
	if cluster.ServiceProviderProperties.ClusterServiceID == nil {
		return utils.TrackError(fmt.Errorf("cluster %s has no ClusterServiceID", cluster.ID))
	}

	// New credential cannot be requested while credentials are being revoked.
	if len(cluster.ServiceProviderProperties.RevokeCredentialsOperationID) > 0 {
		writer.Header().Set("Retry-After", strconv.Itoa(10))
		return coreapi.NewConflictError(clusterResourceID, "Cannot request credential while credentials are being revoked")
	}

	transaction := f.resourcesDBClient.NewTransaction(clusterResourceID.SubscriptionID)

	operationDoc := cosmosstorageutils.NewOperation(
		operationRequest,
		clusterResourceID,
		metadataapi.InternalID{},
		f.azureLocation,
		request.Header.Get(coreapi.HeaderNameHomeTenantID),
		request.Header.Get(coreapi.HeaderNameClientObjectID),
		request.Header.Get(coreapi.HeaderNameAsyncNotificationURI),
		correlationData)
	transaction.OnSuccess(addOperationResponseHeaders(writer, request, operationDoc.NotificationURI, operationDoc.OperationID))
	_, err = f.resourcesDBClient.Operations(clusterResourceID.SubscriptionID).AddCreateToTransaction(ctx, transaction, operationDoc, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	_, err = transaction.Execute(ctx, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	writer.WriteHeader(http.StatusAccepted)
	return nil
}

func (f *Frontend) ArmResourceActionRevokeCredentials(writer http.ResponseWriter, request *http.Request) error {
	const operationRequest = cosmosstorageutils.OperationRequestSystemAdminCredentialRevocation

	ctx := request.Context()
	logger := utils.LoggerFromContext(ctx)

	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	// Parent resource is the hcpOpenShiftCluster.
	clusterResourceID := resourceID.Parent

	correlationData, err := CorrelationDataFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	cluster, err := f.resourcesDBClient.HCPClusters(clusterResourceID.SubscriptionID, clusterResourceID.ResourceGroupName).Get(ctx, clusterResourceID.Name)
	if err != nil {
		return utils.TrackError(err)
	}
	// CheckForProvisioningStateConflict does not log conflict errors
	// but does log unexpected errors like database failures.
	if err := checkForProvisioningStateConflict(ctx, f.resourcesDBClient, operationRequest, cluster.ID, cluster.ServiceProviderProperties.ProvisioningState); err != nil {
		return utils.TrackError(err)
	}
	if cluster.ServiceProviderProperties.ClusterServiceID == nil {
		return utils.TrackError(fmt.Errorf("cluster %s has no ClusterServiceID", cluster.ID))
	}

	// Credential revocation cannot be requested while another revocation is in progress.
	if len(cluster.ServiceProviderProperties.RevokeCredentialsOperationID) > 0 {
		writer.Header().Set("Retry-After", strconv.Itoa(10))
		return coreapi.NewConflictError(clusterResourceID, "Credentials are already being revoked")
	}

	transaction := f.resourcesDBClient.NewTransaction(clusterResourceID.SubscriptionID)

	// Just as deleting an ARM resource cancels any other operations on the resource,
	// revoking credentials cancels any credential requests in progress.
	operationsToCancel, err := corecosmosstorage.CancelActiveOperations(ctx, f.resourcesDBClient, transaction, &corecosmosstorage.ResourcesDBClientListActiveOperationDocsOptions{
		Request:    metadataapi.Ptr(cosmosstorageutils.OperationRequestSystemAdminCredentialRequest),
		ExternalID: clusterResourceID,
	})
	if err != nil {
		return utils.TrackError(err)
	}
	if len(operationsToCancel) > 0 {
		logger.Info("canceling RequestCredential operations", "operationsToCancel", operationsToCancel)
	}

	operationDoc := cosmosstorageutils.NewOperation(
		operationRequest,
		clusterResourceID,
		*cluster.ServiceProviderProperties.ClusterServiceID,
		f.azureLocation,
		request.Header.Get(coreapi.HeaderNameHomeTenantID),
		request.Header.Get(coreapi.HeaderNameClientObjectID),
		request.Header.Get(coreapi.HeaderNameAsyncNotificationURI),
		correlationData)

	transaction.OnSuccess(addOperationResponseHeaders(writer, request, operationDoc.NotificationURI, operationDoc.OperationID))
	_, err = f.resourcesDBClient.Operations(operationDoc.OperationID.SubscriptionID).AddCreateToTransaction(ctx, transaction, operationDoc, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	// Setting this field effectively blocks further credential
	// requests or revocations until the backend clears the field.
	cluster.ServiceProviderProperties.RevokeCredentialsOperationID = operationDoc.OperationID.Name

	_, err = f.resourcesDBClient.HCPClusters(cluster.ID.SubscriptionID, cluster.ID.ResourceGroupName).AddReplaceToTransaction(ctx, transaction, cluster, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	_, err = transaction.Execute(ctx, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	writer.WriteHeader(http.StatusAccepted)
	return nil
}

func (f *Frontend) ArmOperationsList(writer http.ResponseWriter, request *http.Request) error {
	pagedResponse := coreapi.NewPagedResponse()

	for _, operation := range AvailableOperations {
		jsonBytes, err := coreapi.MarshalJSON(operation)
		if err != nil {
			return utils.TrackError(err)
		}
		pagedResponse.AddValue(jsonBytes)
	}

	// XXX We are temporarily hosting an operation list for the
	//     ARO "Classic" service. See routes.go for more context.
	for _, operation := range AvailableClassicOperations {
		jsonBytes, err := coreapi.MarshalJSON(operation)
		if err != nil {
			return utils.TrackError(err)
		}
		pagedResponse.AddValue(jsonBytes)
	}

	_, err := coreapi.WriteJSONResponse(writer, http.StatusOK, pagedResponse)
	if err != nil {
		return utils.TrackError(err)
	}

	return nil
}

func (f *Frontend) ArmSubscriptionGet(writer http.ResponseWriter, request *http.Request) error {
	ctx := request.Context()

	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	subscriptionID := request.PathValue(PathSegmentSubscriptionID)

	subscription, err := f.resourcesDBClient.Subscriptions().Get(ctx, subscriptionID)
	if cosmosstorageutils.IsNotFoundError(err) {
		return coreapi.NewResourceNotFoundError(resourceID)
	}
	if err != nil {
		return utils.TrackError(err)
	}

	_, err = coreapi.WriteJSONResponse(writer, http.StatusOK, subscription)
	if err != nil {
		return utils.TrackError(err)
	}
	return nil
}

func (f *Frontend) ArmSubscriptionPut(writer http.ResponseWriter, request *http.Request) error {
	ctx := request.Context()
	logger := utils.LoggerFromContext(ctx)

	body, err := BodyFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}
	subscriptionID := request.PathValue(PathSegmentSubscriptionID)

	var requestSubscription coreapi.Subscription
	err = json.Unmarshal(body, &requestSubscription)
	if err != nil {
		return coreapi.NewInvalidRequestContentError(err)
	}
	requestSubscription.CosmosMetadata.ResourceID, err = coreapi.ToSubscriptionResourceID(subscriptionID)
	if err != nil {
		return utils.TrackError(err)
	}
	requestSubscription.ResourceID = requestSubscription.CosmosMetadata.ResourceID
	requestSubscription.SetPartitionKey(subscriptionID)

	validationErrs := validation.ValidateSubscriptionCreate(ctx, &requestSubscription)
	if err := coreapi.CloudErrorFromFieldErrors(validationErrs); err != nil {
		return utils.TrackError(err)
	}

	var resultingSubscription *coreapi.Subscription
	existingSubscription, err := f.resourcesDBClient.Subscriptions().Get(ctx, subscriptionID)
	if cosmosstorageutils.IsNotFoundError(err) {
		resultingSubscription, err = f.resourcesDBClient.Subscriptions().Create(ctx, &requestSubscription, nil)
		if err != nil {
			return utils.TrackError(err)
		}
		logger.Info(fmt.Sprintf("created document for subscription %s", subscriptionID))
	} else if err != nil {
		return utils.TrackError(err)
	} else {
		messages := getSubscriptionDifferences(existingSubscription, &requestSubscription)
		for _, message := range messages {
			logger.Info(message)
		}
		if len(messages) > 0 {
			// Carry the etag of the just-read document forward so Replace
			// is conditional on it; the DB layer refuses unconditional updates.
			requestSubscription.CosmosMetadata = *existingSubscription.CosmosMetadata.DeepCopy()
			resultingSubscription, err = f.resourcesDBClient.Subscriptions().Replace(ctx, &requestSubscription, nil)
			if err != nil {
				return utils.TrackError(err)
			}
			logger.Info(fmt.Sprintf("updated document for subscription %s", subscriptionID))
		} else {
			resultingSubscription = existingSubscription
		}
	}

	// Clean up resources if subscription is deleted.
	if resultingSubscription.State == coreapi.SubscriptionStateDeleted {
		if err := f.DeleteAllResourcesInSubscription(ctx, subscriptionID); err != nil {
			return utils.TrackError(err)
		}
	}

	_, err = coreapi.WriteJSONResponse(writer, http.StatusOK, resultingSubscription)
	if err != nil {
		return utils.TrackError(err)
	}
	return nil
}

func (f *Frontend) ArmDeploymentPreflight(writer http.ResponseWriter, request *http.Request) error {
	var subscriptionID = request.PathValue(PathSegmentSubscriptionID)
	var resourceGroup = request.PathValue(PathSegmentResourceGroupName)

	ctx := request.Context()
	logger := utils.LoggerFromContext(ctx)

	subscription, err := SubscriptionFromContext(ctx)
	if err != nil {
		return err
	}

	body, err := BodyFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	// TODO explain why it is safe to decode this directly into an internal type
	deploymentPreflight, err := coreapi.UnmarshalDeploymentPreflight(body)
	if err != nil {
		return utils.TrackError(err)
	}

	preflightErrors := []coreapi.CloudErrorBody{}

	availableAROHCPVersions := f.apiRegistry.ListVersions()
	for index, raw := range deploymentPreflight.Resources {
		var preflightErr error

		// Check the raw JSON for any Template Language Expressions (TLEs).
		// If any are detected, skip the resource because Cluster Service
		// does not handle TLEs in its input validation.
		detectedTLE, err := coreapi.DetectTLE(raw)
		if err != nil {
			preflightErr = coreapi.NewInvalidRequestContentError(err)
			// Preflight is best-effort: a malformed resource is not a validation failure.
			logger.Info("preflight: malformed resource detected", "error", preflightErr.Error())
			continue
		}
		if detectedTLE {
			continue
		}

		preflightResource := &coreapi.DeploymentPreflightResource{}
		err = json.Unmarshal(raw, preflightResource)
		if err != nil {
			preflightErr = coreapi.NewInvalidRequestContentError(err)
			// Preflight is best-effort: a malformed resource is not a validation failure.
			logger.Info("preflight: failed to unmarshal resource", "error", preflightErr.Error())
			continue
		}

		if !availableAROHCPVersions.Has(preflightResource.APIVersion) {
			// Preflight is best-effort: a malformed resource is not a validation failure.
			validationErr := coreapi.CloudErrorBody{
				Code:    coreapi.CloudErrorCodeInvalidRequestContent,
				Message: fmt.Sprintf("Unrecognized API version '%s'", preflightResource.APIVersion),
				Target:  "apiVersion",
			}
			logger.Info(
				fmt.Sprintf("preflight: Resource #%d failed preliminary validation (see details)", index+1),
				"details", validationErr)
			continue
		}

		resourceLogger := logger.WithValues(utils.LogValues{}.AddResourceName(preflightResource.Name).AddResourceType(preflightResource.Type)...)

		switch strings.ToLower(preflightResource.Type) {
		case strings.ToLower(coreapi.ClusterResourceType.String()):
			// API version is already validated by this point.
			versionedInterface, _ := f.apiRegistry.Lookup(preflightResource.APIVersion)
			versionedCluster := versionedInterface.NewHCPOpenShiftCluster(nil)

			err = preflightResource.Convert(versionedCluster)
			if err != nil {
				// Preflight is best effort: failure to parse a resource is not a validation failure.
				resourceLogger.Info("preflight: failed to unmarshal resource", "error", err.Error())
				continue
			}

			newInternalCluster, err := versionedCluster.ConvertToInternal(nil)
			if err != nil {
				resourceLogger.Info("preflight: failed to convert resource", "error", err.Error())
				continue
			}
			// Backstop for fields unknown to this API version's SetDefaultValues*.
			// See docs/api-version-defaults-and-storage.md.
			newInternalCluster.EnsureDefaults()
			newInternalCluster.SystemData = ensureSystemData(newInternalCluster.SystemData, nil)
			// the external type lacks sufficient data to full produce a valid resourceID.  We do that separately here.
			parts := []string{
				"/subscriptions", subscriptionID,
				"resourceGroups", resourceGroup,
				"providers", coreapi.ClusterResourceType.String(), newInternalCluster.Name,
			}
			newInternalCluster.ID, err = azcorearm.ParseResourceID(strings.Join(parts, "/"))
			if err != nil {
				// this indicates something really strange happened, return an error for it.
				return utils.TrackError(err)
			}
			op := operation.Operation{
				Type:    operation.Create,
				Options: []string{validation.ManagedIdentitiesDataPlaneIdentityURLOptionalOperationOption},
			}
			admissionContext, ctxErr := f.newClusterAdmissionContext(ctx, op, subscription, newInternalCluster, nil)
			if ctxErr != nil {
				return utils.TrackError(ctxErr)
			}
			// Apply the same mutations that real cluster creation applies
			if mutationErrs := admission.MutateCluster(ctx, admissionContext, op, newInternalCluster, nil); len(mutationErrs) > 0 {
				preflightErr = coreapi.CloudErrorFromFieldErrors(mutationErrs)
				break
			}
			validationErrs := validation.ValidateCluster(ctx, op, newInternalCluster, nil, metadataapi.Must(versionedInterface.ValidationPathRewriter(&coreapi.HCPOpenShiftCluster{})))
			validationErrs = append(validationErrs, admission.AdmitCluster(ctx, admissionContext, op, newInternalCluster, nil)...)
			preflightErr = coreapi.CloudErrorFromFieldErrors(validationErrs)

		case strings.ToLower(coreapi.NodePoolResourceType.String()):
			// API version is already validated by this point.
			versionedInterface, _ := f.apiRegistry.Lookup(preflightResource.APIVersion)
			versionedNodePool := versionedInterface.NewHCPOpenShiftClusterNodePool(nil)

			err = preflightResource.Convert(versionedNodePool)
			if err != nil {
				// Preflight is best effort: failure to parse a resource is not a validation failure.
				resourceLogger.Info("preflight: failed to unmarshal resource", "error", err.Error())
				continue
			}

			// Perform static validation as if for a node pool creation request.
			newInternalNodePool, err := versionedNodePool.ConvertToInternal(nil)
			if err != nil {
				resourceLogger.Info("preflight: failed to convert resource", "error", err.Error())
				continue
			}
			// Backstop for fields unknown to this API version's SetDefaultValues*.
			// See docs/api-version-defaults-and-storage.md.
			newInternalNodePool.EnsureDefaults()
			// the external type lacks sufficient data to full produce a valid resourceID.  We do that separately here.
			parts := []string{
				"/subscriptions", subscriptionID,
				"resourceGroups", resourceGroup,
				"providers", coreapi.ClusterResourceType.String(), "preflight",
				coreapi.NodePoolResourceType.Types[len(coreapi.NodePoolResourceType.Types)-1], newInternalNodePool.Name,
			}
			newInternalNodePool.ID, err = azcorearm.ParseResourceID(strings.Join(parts, "/"))
			if err != nil {
				// this indicates something really strange happened, return an error for it.
				return utils.TrackError(err)
			}
			newInternalNodePool.SystemData = ensureSystemData(newInternalNodePool.SystemData, nil)

			op := operation.Operation{
				Type:    operation.Create,
				Options: validation.BuildValidationOptions(subscription.GetRegisteredFeatures(), metadataapi.APIVersion(versionedInterface.String())),
			}
			validationErrs := validation.ValidateNodePool(ctx, op, newInternalNodePool, nil)
			preflightErr = coreapi.CloudErrorFromFieldErrors(validationErrs)

		case strings.ToLower(coreapi.ExternalAuthResourceType.String()):
			// API version is already validated by this point.
			versionedInterface, _ := f.apiRegistry.Lookup(preflightResource.APIVersion)
			versionedExternalAuth := versionedInterface.NewHCPOpenShiftClusterExternalAuth(nil)

			err = preflightResource.Convert(versionedExternalAuth)
			if err != nil {
				// Preflight is best effort: failure to parse a resource is not a validation failure.
				resourceLogger.Info("preflight: failed to unmarshal resource", "error", err.Error())
				continue
			}

			// Perform static validation as if for an external auth creation request.
			newInternalAuth, err := versionedExternalAuth.ConvertToInternal(nil)
			if err != nil {
				resourceLogger.Info("preflight: failed to convert resource", "error", err.Error())
				continue
			}
			// Backstop for fields unknown to this API version's SetDefaultValues*.
			// See docs/api-version-defaults-and-storage.md.
			newInternalAuth.EnsureDefaults()
			newInternalAuth.SystemData = ensureSystemData(newInternalAuth.SystemData, nil)
			// the external type lacks sufficient data to full produce a valid resourceID.  We do that separately here.
			parts := []string{
				"/subscriptions", subscriptionID,
				"resourceGroups", resourceGroup,
				"providers", coreapi.ClusterResourceType.String(), "preflight",
				coreapi.ExternalAuthResourceType.Types[len(coreapi.NodePoolResourceType.Types)-1], newInternalAuth.Name,
			}
			newInternalAuth.ID, err = azcorearm.ParseResourceID(strings.Join(parts, "/"))
			if err != nil {
				// this indicates something really strange happened, return an error for it.
				return utils.TrackError(err)
			}
			validationErrs := validation.ValidateExternalAuthCreate(ctx, newInternalAuth)
			preflightErr = coreapi.CloudErrorFromFieldErrors(validationErrs)

		default:
			// Disregard foreign resource types.
			continue
		}

		var cloudError *coreapi.CloudError
		if errors.As(preflightErr, &cloudError) {
			var details []coreapi.CloudErrorBody

			// This avoids double-nesting details when there's multiple errors.
			//
			// To illustrate, instead of:
			//
			// {
			//   "code": "MultipleErrorsOccurred"
			//   "message": "Content validation failed for {{RESOURCE_NAME}}"
			//   "target": "{{RESOURCE_ID}}"
			//   "details": [
			//     {
			//       "code": "MultipleErrorsOccurred"
			//       "message": "Content validation failed on multiple fields"
			//       "details": [
			//         ...field-specific validation errors...
			//       ]
			//     }
			//   ]
			// }
			//
			// we want:
			//
			// {
			//   "code": "MultipleErrorsOccurred"
			//   "message": "Content validation failed for {{RESOURCE_NAME}}"
			//   "target": "{{RESOURCE_ID}}"
			//   "details": [
			//     ...field-specific validation errors...
			//   ]
			// }
			//
			if len(cloudError.Details) > 0 {
				details = cloudError.Details
			} else {
				details = []coreapi.CloudErrorBody{*cloudError.CloudErrorBody}
			}
			preflightErrors = append(preflightErrors, coreapi.CloudErrorBody{
				Code:    cloudError.Code,
				Message: fmt.Sprintf("Content validation failed for '%s'", preflightResource.Name),
				Target:  preflightResource.ResourceID(subscriptionID, resourceGroup),
				Details: details,
			})
			continue
		}

		// FIXME Further preflight steps go here.
	}

	coreapi.WriteDeploymentPreflightResponse(writer, preflightErrors)
	return nil
}

func (f *Frontend) OperationStatus(writer http.ResponseWriter, request *http.Request) error {
	ctx := request.Context()
	logger := utils.LoggerFromContext(ctx)

	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	operation, err := f.resourcesDBClient.Operations(resourceID.SubscriptionID).Get(ctx, resourceID.Name)
	if err != nil {
		return utils.TrackError(err)
	}

	// Validate the identity retrieving the operation result is the
	// same identity that triggered the operation. Return 404 if not.
	if !f.OperationIsVisible(request, operation) {
		logger.Info("operation result not visible to requester")
		writer.WriteHeader(http.StatusNotFound)
		return nil
	}

	_, err = coreapi.WriteJSONResponse(writer, http.StatusOK, cosmosstorageutils.ToStatus(operation))
	if err != nil {
		return utils.TrackError(err)
	}
	return nil
}

func getSubscriptionDifferences(oldSub, newSub *coreapi.Subscription) []string {
	var messages []string

	if oldSub.State != newSub.State {
		messages = append(messages, fmt.Sprintf("Subscription state changed from %s to %s", oldSub.State, newSub.State))
	}

	if oldSub.Properties == nil {
		oldSub.Properties = &coreapi.SubscriptionProperties{}
	}
	if newSub.Properties == nil {
		newSub.Properties = &coreapi.SubscriptionProperties{}
	}

	var oldTenantId, newTenantId string

	if oldSub.Properties.TenantId != nil {
		oldTenantId = *oldSub.Properties.TenantId
	}
	if newSub.Properties.TenantId != nil {
		newTenantId = *newSub.Properties.TenantId
	}

	if oldTenantId != newTenantId {
		messages = append(messages, fmt.Sprintf("Subscription tenantId changed from %s to %s", oldTenantId, newTenantId))
	}

	oldFeatures := featuresMap(oldSub.Properties.RegisteredFeatures)
	newFeatures := featuresMap(newSub.Properties.RegisteredFeatures)

	for featureName, oldState := range oldFeatures {
		newState, exists := newFeatures[featureName]
		if !exists {
			messages = append(messages, fmt.Sprintf("Feature %s removed", featureName))
		} else if oldState != newState {
			messages = append(messages, fmt.Sprintf("Feature %s state changed from %s to %s", featureName, oldState, newState))
		}
	}
	for featureName, newState := range newFeatures {
		if _, exists := oldFeatures[featureName]; !exists {
			messages = append(messages, fmt.Sprintf("Feature %s added with state %s", featureName, newState))
		}
	}

	return messages
}

func (f *Frontend) OperationResult(writer http.ResponseWriter, request *http.Request) error {
	ctx := request.Context()

	versionedInterface, err := VersionFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	operation, err := f.resourcesDBClient.Operations(resourceID.SubscriptionID).Get(ctx, resourceID.Name)
	if err != nil {
		return utils.TrackError(err)
	}

	// Validate the identity retrieving the operation result is the
	// same identity that triggered the operation. Return 404 if not.
	if !f.OperationIsVisible(request, operation) {
		return coreapi.NewResourceNotFoundError(resourceID)
	}

	// Handle non-terminal statuses and (maybe?) failure/cancellation.
	//
	// XXX ARM requirements for failed async operations get fuzzy here.
	//
	//     My best understanding, based on a Stack Overflow answer [1], is
	//     returning an Azure-AsyncOperation header will cause ARM to poll
	//     that endpoint first.
	//
	//     If ARM finds the operation in a "Failed" or "Canceled" state,
	//     it will propagate details from the response's "error" property.
	//
	//     If ARM finds the operation in a "Succeeded" state, ONLY THEN is
	//     this endpoint called (if a Location header was also returned).
	//
	//     So for the "Failed or Canceled" case we just give a generic
	//     "Internal Server Error" response since, in theory, this case
	//     should never be reached.
	//
	//     [1] https://stackoverflow.microsoft.com/a/318573/106707
	//
	switch operation.Status {
	case coreapi.ProvisioningStateSucceeded:
		// Handled below.
	case coreapi.ProvisioningStateFailed, coreapi.ProvisioningStateCanceled:
		return fmt.Errorf("invalid operation status: %s", operation.Status)
	default:
		// Operation is still in progress.
		AddLocationHeader(writer, request, operation.OperationID)
		writer.WriteHeader(http.StatusAccepted)
		return nil
	}

	// The response henceforth should be exactly as though the operation
	// succeeded synchronously.

	var successStatusCode int

	switch operation.Request {
	case cosmosstorageutils.OperationRequestCreate:
		successStatusCode = http.StatusCreated
	case cosmosstorageutils.OperationRequestUpdate:
		successStatusCode = http.StatusOK
	case cosmosstorageutils.OperationRequestDelete:
		writer.WriteHeader(http.StatusNoContent)
		return nil
	case cosmosstorageutils.OperationRequestSystemAdminCredentialRequest:
		successStatusCode = http.StatusOK
	case cosmosstorageutils.OperationRequestSystemAdminCredentialRevocation:
		writer.WriteHeader(http.StatusNoContent)
		return nil
	default:
		return fmt.Errorf("unhandled request type: %s", operation.Request)
	}

	var responseBody []byte

	switch {
	case operation.InternalID.Kind() == cmv1.BreakGlassCredentialKind:
		csBreakGlassCredential, err := f.clusterServiceClient.GetBreakGlassCredential(ctx, operation.InternalID)
		if err != nil {
			return utils.TrackError(err)
		}

		responseBody, err = versionedInterface.MarshalHCPOpenShiftClusterAdminCredential(ocm.ConvertCStoAdminCredential(csBreakGlassCredential))
		if err != nil {
			return utils.TrackError(err)
		}

	case armhelpers.ResourceTypeEqual(operation.ExternalID.ResourceType, coreapi.ClusterResourceType):
		resultingInternalCluster, err := f.getInternalClusterFromStorage(ctx, operation.ExternalID)
		if err != nil {
			return utils.TrackError(err)
		}
		responseBody, err = coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftCluster(resultingInternalCluster))
		if err != nil {
			return utils.TrackError(err)
		}

	case operation.ExternalID.ResourceType.String() == coreapi.NodePoolResourceType.String():
		resultingInternalNodePool, err := f.getInternalNodePoolFromStorage(ctx, operation.ExternalID)
		if err != nil {
			return utils.TrackError(err)
		}
		responseBody, err = coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftClusterNodePool(resultingInternalNodePool))
		if err != nil {
			return utils.TrackError(err)
		}

	case operation.ExternalID.ResourceType.String() == coreapi.ExternalAuthResourceType.String():
		resultingInternalExternalAuth, err := f.getInternalExternalAuthFromStorage(ctx, operation.ExternalID)
		if err != nil {
			return utils.TrackError(err)
		}
		responseBody, err = coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftClusterExternalAuth(resultingInternalExternalAuth))
		if err != nil {
			return utils.TrackError(err)
		}

	default:
		return fmt.Errorf("unsupported operation reference: %s", operation.ExternalID)
	}

	_, err = coreapi.WriteJSONResponse(writer, successStatusCode, responseBody)
	if err != nil {
		return utils.TrackError(err)
	}
	return nil
}

func featuresMap(features *[]coreapi.Feature) map[string]string {
	featureMap := make(map[string]string)
	if features != nil {
		for _, feature := range *features {
			if feature.Name != nil && feature.State != nil {
				featureMap[*feature.Name] = *feature.State
			}
		}
	}
	return featureMap
}

func marshalCSVersion(resourceID *azcorearm.ResourceID, version *arohcpv1alpha1.Version, versionedInterface coreapi.Version) ([]byte, error) {
	hcpVersion := ocm.ConvertCStoHCPOpenShiftVersion(resourceID, version)
	return coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftVersion(hcpVersion))
}
