// 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"
	"fmt"
	"maps"
	"net/http"
	"strings"
	"time"

	"k8s.io/apimachinery/pkg/api/operation"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/utils/ptr"

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

	"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/conversion"
	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/corecosmosstorage"
	"github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils"
	"github.com/Azure/ARO-HCP/internal/utils"
	"github.com/Azure/ARO-HCP/internal/validation"
)

// GetHCPCluster implements the GET single resource API contract for HCP Clusters
// * 200 If the resource exists
// * 404 If the resource does not exist
func (f *Frontend) GetHCPCluster(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) // used for error reporting
	if err != nil {
		return utils.TrackError(err)
	}

	resultingInternalCluster, err := f.getInternalClusterFromStorage(ctx, resourceID)
	if err != nil {
		return utils.TrackError(err)
	}
	responseBytes, err := coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftCluster(resultingInternalCluster))
	if err != nil {
		return utils.TrackError(err)
	}

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

	return nil
}

func (f *Frontend) ArmResourceListClusters(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)
	resourceGroupName := request.PathValue(PathSegmentResourceGroupName)

	pagedResponse := coreapi.NewPagedResponse()

	// Cluster list is served entirely from Cosmos DB. Cosmos's continuation token also meets
	// the requirements of a skipToken for ARM pagination, so it is used directly as the
	// nextLink token below.

	internalClusterIterator, err := f.resourcesDBClient.HCPClusters(subscriptionID, resourceGroupName).List(ctx, dbListOptionsFromRequest(request))
	if err != nil {
		return utils.TrackError(err)
	}
	for _, internalCluster := range internalClusterIterator.Items(ctx) {
		resultingExternalCluster := versionedInterface.NewHCPOpenShiftCluster(internalCluster)
		jsonBytes, err := coreapi.MarshalJSON(resultingExternalCluster)
		if err != nil {
			return utils.TrackError(err)
		}
		pagedResponse.AddValue(jsonBytes)
	}
	err = internalClusterIterator.GetError()
	if err != nil {
		return utils.TrackError(err)
	}
	// MiddlewareReferer ensures Referer is present.
	err = pagedResponse.SetNextLink(request.Referer(), internalClusterIterator.GetContinuationToken())
	if err != nil {
		return utils.TrackError(err)
	}

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

	return nil
}

// CreateOrUpdateHCPCluster handles both PUT and PATCH requests from clients to ARM. The semantics of these verbs are:
// When no resource by this identifier exists:
//   - PUT: create the resource, overlaying client-specified fields on a default resource struct,
//     which only has API-specified non-zero default values
//   - PATCH: error
//
// When a resource by this identifier already exists:
//   - PUT: replace the resource, overlaying client-specified fields on a default resource struct,
//     which only has API-specified non-zero default values (and NOT taking into account the
//     existing content of the resource in storage)
//   - PATCH: update the resource, overlaying client-specified fields onto the existing resource
//
// This means all required properties must be specified in the request body for PUT, whether creating or updating a resource.
//
// See these documents for a description of required semantics from ARM Resource Providers:
//
// PUT:
// The Put operation creates or replaces a resource. Resource types can be nested and, if so, must follow the Resource API guidelines.
// ARM does not distinguish between creation and replacement requests. The resource provider should consult its datastore if a distinction
// is necessary. A Put should always be allowed to overwrite an existing resource. An exception to this rule is when
// `x-ms-mutability ['create', 'read']` is applied to a property. For such properties, the value submitted during the create operation
// must be retained. Eg: The location property defined for tracked resources.
// ref: https://github.com/cloud-and-ai-microsoft/resource-provider-contract/blob/master/v1.0/put-resource.md#put-resource
//
// PATCH
// A Patch operation updates a resource and it is strongly recommended that service teams implement it. ARM requires RPs to support Patch for updating tags for a tracked resource.
// ref: https://github.com/cloud-and-ai-microsoft/resource-provider-contract/blob/master/v1.0/patch-resource.md#patch-resource
func (f *Frontend) CreateOrUpdateHCPCluster(writer http.ResponseWriter, request *http.Request) error {
	var err error

	ctx := request.Context()

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

	oldInternalCluster, err := f.resourcesDBClient.HCPClusters(resourceID.SubscriptionID, resourceID.ResourceGroupName).Get(ctx, resourceID.Name)
	if err != nil && !cosmosstorageutils.IsNotFoundError(err) {
		return utils.TrackError(err)
	}

	updating := oldInternalCluster != nil
	if updating {
		// CheckForProvisioningStateConflict does not log conflict errors
		// but does log unexpected errors like database failures.
		if err := checkForProvisioningStateConflict(ctx, f.resourcesDBClient, cosmosstorageutils.OperationRequestUpdate, oldInternalCluster.ID, oldInternalCluster.ServiceProviderProperties.ProvisioningState); err != nil {
			return utils.TrackError(err)
		}

		switch request.Method {
		case http.MethodPut:
			return f.updateHCPCluster(writer, request, oldInternalCluster)
		case http.MethodPatch:
			return f.patchHCPCluster(writer, request, oldInternalCluster)
		default:
			return fmt.Errorf("unsupported method %s", request.Method)
		}
	}

	switch request.Method {
	case http.MethodPut:
		return f.createHCPCluster(writer, request)
	case http.MethodPatch:
		return coreapi.NewResourceNotFoundError(resourceID)
	default:
		return fmt.Errorf("unsupported method %s", request.Method)
	}
}

func decodeDesiredClusterCreate(ctx context.Context, azureLocation string, requestHeader http.Header) (*coreapi.HCPOpenShiftCluster, error) {
	versionedInterface, err := VersionFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	body, err := BodyFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	systemData, err := SystemDataFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	// If for some reason systemData.CreatedAt is not set, we set it to the current time in UTC.
	if systemData.CreatedAt == nil {
		systemData.CreatedAt = ptr.To(time.Now().UTC())
	}

	externalClusterFromRequest := versionedInterface.NewHCPOpenShiftCluster(nil)
	if err := json.Unmarshal(body, &externalClusterFromRequest); err != nil {
		return nil, utils.TrackError(err)
	}
	newInternalCluster, err := externalClusterFromRequest.ConvertToInternal(nil)
	if err != nil {
		return nil, utils.TrackError(err)
	}

	// Backstop for fields unknown to this API version's SetDefaultValues*.
	// See docs/api-version-defaults-and-storage.md.
	newInternalCluster.EnsureDefaults()
	if len(newInternalCluster.Name) > 0 && newInternalCluster.Name != resourceID.Name {
		return nil, nameResourceIDMismatch(resourceID, newInternalCluster.Name)
	}
	// TrackedResource info doesn't appear to come from the external resource information
	conversion.CopyReadOnlyTrackedResourceValues(&newInternalCluster.TrackedResource, ptr.To(coreapi.NewTrackedResource(resourceID, azureLocation)))
	newInternalCluster.SetResourceID(resourceID)
	newInternalCluster.SetPartitionKey(resourceID.SubscriptionID)

	// set fields that were not included during the conversion, because the user does not provide them or because the
	// data is determined live on read.
	newInternalCluster.SystemData = ensureSystemData(systemData, nil)

	// We set the managed identities data plane identity URL associated to the cluster from the
	// http header 'X-Ms-Identity-Url'.
	newInternalCluster.ServiceProviderProperties.ManagedIdentitiesDataPlaneIdentityURL = requestHeader.Get(coreapi.HeaderNameIdentityURL)

	return newInternalCluster, nil
}

// newClusterAdmissionContext creates an admission context for cluster operations.
// The subscription is always required. originalCluster is the inbound cluster as
// the user submitted it — admission deepcopies it to use as a read-only source
// for fields (like tags) that earlier mutations may overwrite on the live
// cluster. For UPDATE operations the caller must also pass the cluster's
// existing resource ID; the constructor then prefetches the
// ServiceProviderCluster and the list of node pools (plus their service-provider
// records) so admission can validate version skew without hitting the DB itself.
// On CREATE pass a nil clusterResourceID — no prior state exists to prefetch.
func (f *Frontend) newClusterAdmissionContext(ctx context.Context, op operation.Operation, subscription *coreapi.Subscription, originalCluster *coreapi.HCPOpenShiftCluster, clusterResourceID *azcorearm.ResourceID) (*admission.ClusterAdmissionContext, error) {
	if subscription == nil {
		return nil, fmt.Errorf("subscription is required for admission context")
	}
	if originalCluster == nil {
		return nil, fmt.Errorf("originalCluster is required for admission context")
	}

	admissionContext := &admission.ClusterAdmissionContext{
		Clock:           f.clock,
		Subscription:    subscription,
		OriginalCluster: originalCluster.DeepCopy(),
	}

	if op.Type == operation.Create {
		subscriptionID := originalCluster.ID.SubscriptionID
		clusterIterator, err := f.resourcesDBClient.HCPClusters(subscriptionID, "").List(ctx, nil)
		if err != nil {
			return nil, fmt.Errorf("cannot list clusters for cluster admission: %w", err)
		}
		for _, cluster := range clusterIterator.Items(ctx) {
			admissionContext.SubscriptionClusters = append(admissionContext.SubscriptionClusters, cluster)

			nodePoolIterator, err := f.resourcesDBClient.HCPClusters(subscriptionID, cluster.ID.ResourceGroupName).NodePools(cluster.ID.Name).List(ctx, nil)
			if err != nil {
				return nil, fmt.Errorf("cannot list node pools for cluster admission: %w", err)
			}
			for _, nodePool := range nodePoolIterator.Items(ctx) {
				admissionContext.SubscriptionNodePools = append(admissionContext.SubscriptionNodePools, nodePool)
			}
			if err := nodePoolIterator.GetError(); err != nil {
				return nil, fmt.Errorf("cannot list node pools for cluster admission: %w", err)
			}
		}
		if err := clusterIterator.GetError(); err != nil {
			return nil, fmt.Errorf("cannot list clusters for cluster admission: %w", err)
		}
		return admissionContext, nil
	}

	if op.Type != operation.Update {
		return admissionContext, nil
	}

	if clusterResourceID == nil {
		return nil, fmt.Errorf("clusterResourceID is required for UPDATE operations")
	}

	spCluster, err := corecosmosstorage.GetOrCreateServiceProviderCluster(ctx, f.resourcesDBClient, clusterResourceID)
	if err != nil {
		return nil, err
	}
	admissionContext.ServiceProviderCluster = spCluster

	nodePoolIterator, err := f.resourcesDBClient.HCPClusters(clusterResourceID.SubscriptionID, clusterResourceID.ResourceGroupName).NodePools(clusterResourceID.Name).List(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("cannot list node pools for cluster admission: %w", err)
	}
	for _, nodePool := range nodePoolIterator.Items(ctx) {
		spNodePool, err := corecosmosstorage.GetOrCreateServiceProviderNodePool(ctx, f.resourcesDBClient, nodePool.ID)
		if err != nil {
			return nil, fmt.Errorf("cannot load service provider node pool %s: %w", nodePool.ID, err)
		}
		admissionContext.ClusterNodePools = append(admissionContext.ClusterNodePools, admission.ClusterAdmissionNodePool{
			NodePool:                nodePool,
			ServiceProviderNodePool: spNodePool,
		})
	}
	if err := nodePoolIterator.GetError(); err != nil {
		return nil, fmt.Errorf("cannot list node pools for cluster admission: %w", err)
	}

	return admissionContext, nil
}

func (f *Frontend) createHCPCluster(writer http.ResponseWriter, request *http.Request) error {
	var err error

	// This handles both PUT and PATCH requests. PATCH requests will
	// never create a new resource. The only other notable difference
	// is the target struct that request bodies are overlayed onto:
	//
	// PUT requests overlay the request body onto a default resource
	// struct, which only has API-specified non-zero default values.
	// This means all required properties must be specified in the
	// request body, whether creating or updating a resource.
	//
	// PATCH requests overlay the request body onto a resource struct
	// that represents an existing resource to be updated.

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

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

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

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

	newInternalCluster, err := decodeDesiredClusterCreate(ctx, f.azureLocation, request.Header)
	if err != nil {
		return utils.TrackError(err)
	}

	validationOp := operation.Operation{
		Type:    operation.Create,
		Options: validation.BuildValidationOptions(subscription.GetRegisteredFeatures(), metadataapi.APIVersion(versionedInterface.String())),
	}
	admissionContext, err := f.newClusterAdmissionContext(ctx, validationOp, subscription, newInternalCluster, nil)
	if err != nil {
		return utils.TrackError(err)
	}
	if mutationErrs := admission.MutateCluster(ctx, admissionContext, validationOp, newInternalCluster, nil); len(mutationErrs) > 0 {
		return utils.TrackError(coreapi.CloudErrorFromFieldErrors(mutationErrs))
	}
	validationErrs := validation.ValidateCluster(ctx, validationOp, newInternalCluster, nil, metadataapi.Must(versionedInterface.ValidationPathRewriter(&coreapi.HCPOpenShiftCluster{})))
	validationErrs = append(validationErrs, admission.AdmitCluster(ctx, admissionContext, validationOp, newInternalCluster, nil)...)
	if err := coreapi.CloudErrorFromFieldErrors(validationErrs); err != nil {
		return utils.TrackError(err)
	}

	// we must validate using user provided .Identity.UserAssignedIdentities because that is the intent expressed by the user to allow
	// us to use these identities. The information contained in those key is not trusted to be accurate, so we clear this field and set to
	// a valid, but empty set of information
	if newInternalCluster.Identity != nil {
		newInternalCluster.Identity.UserAssignedIdentities = nil
	}
	completeClusterIdentity(newInternalCluster, nil)

	logger.Info(fmt.Sprintf("creating resource %s", newInternalCluster.ID))

	transaction := f.resourcesDBClient.NewTransaction(newInternalCluster.ID.SubscriptionID)

	// TODO extract to straight instance creation and then validation.
	clusterCreateOperation := cosmosstorageutils.NewOperation(
		cosmosstorageutils.OperationRequestCreate,
		newInternalCluster.ID,
		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, clusterCreateOperation.NotificationURI, clusterCreateOperation.OperationID))
	_, err = f.resourcesDBClient.Operations(newInternalCluster.ID.SubscriptionID).AddCreateToTransaction(ctx, transaction, clusterCreateOperation, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	// set fields that were not known until the operation doc instance was created.
	// TODO once we we have separate creation/validation of operation documents, this can be done ahead of time.
	newInternalCluster.ServiceProviderProperties.ActiveOperationID = clusterCreateOperation.ResourceID.Name
	newInternalCluster.ServiceProviderProperties.ProvisioningState = clusterCreateOperation.Status

	cosmosUID, err := f.resourcesDBClient.HCPClusters(newInternalCluster.ID.SubscriptionID, newInternalCluster.ID.ResourceGroupName).AddCreateToTransaction(ctx, transaction, newInternalCluster, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	transactionResult, err := transaction.Execute(ctx, &azcosmos.TransactionalBatchOptions{
		EnableContentResponseOnWrite: true,
	})
	if err != nil {
		return utils.TrackError(err)
	}

	// Read back the resource document so the response body is accurate.
	resultingUncastInternalCluster, err := transactionResult.GetItem(cosmosUID)
	if err != nil {
		return utils.TrackError(err)
	}
	resultingInternalCluster, ok := resultingUncastInternalCluster.(*coreapi.HCPOpenShiftCluster)
	if !ok {
		return fmt.Errorf("unexpected type %T", resultingUncastInternalCluster)
	}

	responseBytes, err := coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftCluster(resultingInternalCluster))
	if err != nil {
		return utils.TrackError(err)
	}

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

func decodeDesiredClusterReplace(ctx context.Context, oldInternalCluster *coreapi.HCPOpenShiftCluster) (*coreapi.HCPOpenShiftCluster, error) {
	versionedInterface, err := VersionFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}

	// Decoding for update has a series of semantics for determining the final desired update
	// 1. exact user request
	// 2. defaults for that version
	// 3. if not set, the values that the user doesn't necessary have to set but are not static defaults.  These from from the old value.
	// 4. values that are missing because the external type doesn't represent them
	// 5. values that might change because our machinery changes them.

	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	body, err := BodyFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	systemData, err := SystemDataFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	// Exact user request
	externalClusterFromRequest := versionedInterface.NewHCPOpenShiftCluster(nil)
	if err := json.Unmarshal(body, &externalClusterFromRequest); err != nil {
		return nil, utils.TrackError(err)
	}

	newInternalCluster, err := externalClusterFromRequest.ConvertToInternal(oldInternalCluster)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	if len(newInternalCluster.Name) > 0 && newInternalCluster.Name != resourceID.Name {
		return nil, nameResourceIDMismatch(resourceID, newInternalCluster.Name)
	}

	// values a user doesn't have to provide, but are not static defaults (set dynamically during create).  Set these from old value
	if len(newInternalCluster.CustomerProperties.DNS.BaseDomainPrefix) == 0 {
		newInternalCluster.CustomerProperties.DNS.BaseDomainPrefix = oldInternalCluster.CustomerProperties.DNS.BaseDomainPrefix
	}
	if len(newInternalCluster.CustomerProperties.Platform.ManagedResourceGroup) == 0 {
		newInternalCluster.CustomerProperties.Platform.ManagedResourceGroup = oldInternalCluster.CustomerProperties.Platform.ManagedResourceGroup
	}

	// ServiceProviderProperties contains two types of information
	// 1. values that a user cannot change because the external type does not expose the information.
	//    We must overwrite those values with the oldInternalCluster values so the values don't change, because the user's input will always be empty.
	// 2. values that a user cannot change due to validation requirements, but the user *can* specify the values.
	//    We are overwriting these values that we consider to be status values.
	//    We do this because if a user has read a value, then modified it, then replaces it, we don't want to produce
	//    validation errors on status fields that the user isn't trying to modify.
	conversion.CopyReadOnlyClusterValues(newInternalCluster, oldInternalCluster)
	newInternalCluster.SystemData = ensureSystemData(systemData, oldInternalCluster.SystemData)

	// Here the difference between a nil map and an empty map is significant.
	// If the Tags map is nil, that means it was omitted from the request body,
	// so we leave any existing tags alone. If the Tags map is non-nil, even if
	// empty, that means it was specified in the request body and should fully
	// replace any existing tags.
	if newInternalCluster.Tags == nil {
		newInternalCluster.Tags = maps.Clone(oldInternalCluster.Tags)
	}

	// set any missing defaults
	newInternalCluster.EnsureDefaults()

	return newInternalCluster, nil
}

func (f *Frontend) updateHCPCluster(writer http.ResponseWriter, request *http.Request, oldInternalCluster *coreapi.HCPOpenShiftCluster) error {
	// PUT requests overlay the request body onto a default resource
	// struct, which only has API-specified non-zero default values.
	// This means all required properties must be specified in the
	// request body, whether creating or updating a resource.

	ctx := request.Context()

	newInternalCluster, err := decodeDesiredClusterReplace(ctx, oldInternalCluster)
	if err != nil {
		return utils.TrackError(err)
	}

	return f.updateHCPClusterInCosmos(ctx, writer, request, http.StatusOK, newInternalCluster, oldInternalCluster)
}

func decodeDesiredClusterPatch(ctx context.Context, oldInternalCluster *coreapi.HCPOpenShiftCluster) (*coreapi.HCPOpenShiftCluster, error) {
	versionedInterface, err := VersionFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	resourceID, err := utils.ResourceIDFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	body, err := BodyFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	systemData, err := SystemDataFromContext(ctx)
	if err != nil {
		return nil, utils.TrackError(err)
	}

	// TODO find a way to represent the desired change without starting from internal state here (very confusing)
	// TODO we appear to lack a test, but this seems to take an original, apply the patch and unmarshal the result, meaning the above patch step is just incorrect.
	var newExternalCluster = versionedInterface.NewHCPOpenShiftCluster(oldInternalCluster)
	if err := coreapi.ApplyRequestBody(http.MethodPatch, body, newExternalCluster); err != nil {
		return nil, utils.TrackError(err)
	}
	newInternalCluster, err := newExternalCluster.ConvertToInternal(oldInternalCluster)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	if len(newInternalCluster.Name) > 0 && newInternalCluster.Name != resourceID.Name {
		return nil, nameResourceIDMismatch(resourceID, newInternalCluster.Name)
	}

	// ServiceProviderProperties contains two types of information
	// 1. values that a user cannot change because the external type does not expose the information.
	//    We must overwrite those values with the oldInternalCluster values so the values don't change, because the user's input will always be empty.
	// 2. values that a user cannot change due to validation requirements, but the user *can* specify the values.
	//    We are overwriting these values that we consider to be status values.
	//    We do this because if a user has read a value, then modified it, then replaces it, we don't want to produce
	//    validation errors on status fields that the user isn't trying to modify.
	conversion.CopyReadOnlyClusterValues(newInternalCluster, oldInternalCluster)
	newInternalCluster.SystemData = ensureSystemData(systemData, oldInternalCluster.SystemData)

	// Here the difference between a nil map and an empty map is significant.
	// If the Tags map is nil, that means it was omitted from the request body,
	// so we leave any existing tags alone. If the Tags map is non-nil, even if
	// empty, that means it was specified in the request body and should fully
	// replace any existing tags.
	if newInternalCluster.Tags == nil {
		newInternalCluster.Tags = maps.Clone(oldInternalCluster.Tags)
	}

	// set any missing defaults
	newInternalCluster.EnsureDefaults()

	return newInternalCluster, nil
}

func (f *Frontend) patchHCPCluster(writer http.ResponseWriter, request *http.Request, oldInternalCluster *coreapi.HCPOpenShiftCluster) error {
	// PATCH requests overlay the request body onto a resource struct
	// that represents an existing resource to be updated.
	ctx := request.Context()

	newInternalCluster, err := decodeDesiredClusterPatch(ctx, oldInternalCluster)
	if err != nil {
		return utils.TrackError(err)
	}

	return f.updateHCPClusterInCosmos(ctx, writer, request, http.StatusAccepted, newInternalCluster, oldInternalCluster)
}

func (f *Frontend) updateHCPClusterInCosmos(ctx context.Context, writer http.ResponseWriter, request *http.Request, httpStatusCode int, newInternalCluster, oldInternalCluster *coreapi.HCPOpenShiftCluster) error {
	subscription, err := f.resourcesDBClient.Subscriptions().Get(ctx, oldInternalCluster.ID.SubscriptionID)
	if err != nil {
		return utils.TrackError(err)
	}

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

	validationOp := operation.Operation{
		Type:    operation.Update,
		Options: validation.BuildValidationOptions(subscription.GetRegisteredFeatures(), metadataapi.APIVersion(versionedInterface.String())),
	}
	admissionContext, err := f.newClusterAdmissionContext(ctx, validationOp, subscription, newInternalCluster, oldInternalCluster.ID)
	if err != nil {
		return utils.TrackError(err)
	}
	if mutationErrs := admission.MutateCluster(ctx, admissionContext, validationOp, newInternalCluster, oldInternalCluster); len(mutationErrs) > 0 {
		return utils.TrackError(coreapi.CloudErrorFromFieldErrors(mutationErrs))
	}

	validationErrs := validation.ValidateCluster(ctx, validationOp, newInternalCluster, oldInternalCluster, metadataapi.Must(versionedInterface.ValidationPathRewriter(&coreapi.HCPOpenShiftCluster{})))
	validationErrs = append(validationErrs, admission.AdmitCluster(ctx, admissionContext, validationOp, newInternalCluster, oldInternalCluster)...)
	if err := coreapi.CloudErrorFromFieldErrors(validationErrs); err != nil {
		return utils.TrackError(err)
	}

	// we must validate using user provided .Identity.UserAssignedIdentities because that is the intent expressed by the user to allow
	// us to use these identities. The information contained in those key is not trusted to be accurate, so we clear this field and set to
	// a valid, but empty set of information
	if newInternalCluster.Identity != nil {
		newInternalCluster.Identity.UserAssignedIdentities = nil
	}
	var existingUserAssignedIdentities map[string]*coreapi.UserAssignedIdentity
	if oldInternalCluster.Identity != nil {
		existingUserAssignedIdentities = oldInternalCluster.Identity.UserAssignedIdentities
	}
	completeClusterIdentity(newInternalCluster, existingUserAssignedIdentities)

	transaction := f.resourcesDBClient.NewTransaction(oldInternalCluster.ID.SubscriptionID)
	clusterUpdateOperation := cosmosstorageutils.NewOperation(
		cosmosstorageutils.OperationRequestUpdate,
		oldInternalCluster.ID,
		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, clusterUpdateOperation.NotificationURI, clusterUpdateOperation.OperationID))
	_, err = f.resourcesDBClient.Operations(newInternalCluster.ID.SubscriptionID).AddCreateToTransaction(ctx, transaction, clusterUpdateOperation, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	// set fields that were not known until the operation doc instance was created.
	// TODO once we we have separate creation/validation of operation documents, this can be done ahead of time.
	newInternalCluster.ServiceProviderProperties.ActiveOperationID = clusterUpdateOperation.ResourceID.Name
	newInternalCluster.ServiceProviderProperties.ProvisioningState = clusterUpdateOperation.Status

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

	transactionResult, err := transaction.Execute(ctx, &azcosmos.TransactionalBatchOptions{
		EnableContentResponseOnWrite: true,
	})
	if err != nil {
		return utils.TrackError(err)
	}

	// Read back the resource document so the response body is accurate.
	resultingUncastObj, err := transactionResult.GetItem(oldInternalCluster.GetCosmosData().GetCosmosUID())
	if err != nil {
		return utils.TrackError(err)
	}
	resultingInternalCluster := resultingUncastObj.(*coreapi.HCPOpenShiftCluster)

	responseBytes, err := coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftCluster(resultingInternalCluster))
	if err != nil {
		return utils.TrackError(err)
	}

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

// DeleteCluster implements the deletion API contract for ARM
// * 200 if a deletion is successful
// * 202 if an asynchronous delete is initiated
// * 204 if a well-formed request attempts to delete a nonexistent resource
func (f *Frontend) DeleteCluster(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)
	}

	cluster, err := f.resourcesDBClient.HCPClusters(resourceID.SubscriptionID, resourceID.ResourceGroupName).Get(ctx, resourceID.Name)
	if cosmosstorageutils.IsNotFoundError(err) {
		// For resource not found errors on deletion, ARM requires
		writer.WriteHeader(http.StatusNoContent)
		return nil
	}
	if err != nil {
		return utils.TrackError(err)
	}

	if err := checkForProvisioningStateConflict(ctx, f.resourcesDBClient, cosmosstorageutils.OperationRequestDelete, cluster.ID, cluster.ServiceProviderProperties.ProvisioningState); err != nil {
		return utils.TrackError(err)
	}

	logger.Info(fmt.Sprintf("deleting resource %s", cluster.ID))

	transaction := f.resourcesDBClient.NewTransaction(cluster.ID.SubscriptionID)
	if err := f.addDeleteClusterToTransaction(ctx, writer, request, transaction, cluster); err != nil {
		return utils.TrackError(err)
	}
	_, err = transaction.Execute(ctx, nil)
	if err != nil {
		logger.Error(err, "failed executing transaction", "transaction", transaction)
		return utils.TrackError(err)
	}

	writer.WriteHeader(http.StatusAccepted)
	return nil
}

func (f *Frontend) addDeleteClusterToTransaction(ctx context.Context, writer http.ResponseWriter, request *http.Request, transaction cosmosstorageutils.DBTransaction, cluster *coreapi.HCPOpenShiftCluster) error {
	correlationData, err := CorrelationDataFromContext(ctx)
	if err != nil {
		return utils.TrackError(err)
	}

	// Cluster Service will take care of canceling any ongoing operations
	// on the resource or child resources, but we need to do some database
	// bookkeeping to reflect that.
	_, err = corecosmosstorage.CancelActiveOperations(ctx, f.resourcesDBClient, transaction, &corecosmosstorage.ResourcesDBClientListActiveOperationDocsOptions{
		ExternalID: cluster.ID,
		// We don't include operations for resources below clusters (eg. nodepools) because, as part of the deletion flow,
		// we will process each nested resource directly, delete it and cancel its operations then. If we handle resources
		// multiple times, then we end up in a situation where the first operation to cancel operations has a matching etag
		// and the subsequent operations do not have a matching etag, meaning we fail the transaction.
		IncludeNestedResources: false,
	})
	if err != nil {
		return utils.TrackError(err)
	}

	operationDoc := cosmosstorageutils.NewOperation(
		cosmosstorageutils.OperationRequestDelete,
		cluster.ID,
		metadataapi.InternalID{},
		f.azureLocation,
		"",
		"",
		"",
		correlationData)
	// TODO remove this once migration of the new cluster deletion from frontend to backend approach is fully completed in all ARO-HCP
	// permanent environments, for all regions.
	operationDoc.UsesNewClusterDeletionApproach = true
	if request != nil {
		// these are optional because when this is triggered via the subscription deletion flow, there is no
		// deletion request containing these headers so these operations cannot be directly tracked.
		operationDoc.TenantID = request.Header.Get(coreapi.HeaderNameHomeTenantID)
		operationDoc.ClientID = request.Header.Get(coreapi.HeaderNameClientObjectID)
		operationDoc.NotificationURI = request.Header.Get(coreapi.HeaderNameAsyncNotificationURI)
		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)
	}

	if cluster.ServiceProviderProperties.DeletionTimestamp == nil {
		cluster.ServiceProviderProperties.DeletionTimestamp = &metav1.Time{Time: f.clock.Now().UTC()}
	}
	cluster.ServiceProviderProperties.ActiveOperationID = operationDoc.ResourceID.Name
	cluster.ServiceProviderProperties.ProvisioningState = operationDoc.Status
	// TODO remove this once migration of the new cluster deletion from frontend to backend approach is fully completed in all ARO-HCP
	// permanent environments, for all regions.
	cluster.ServiceProviderProperties.UsesNewClusterDeletionApproach = true

	cluster.ServiceProviderProperties.DeleteOperationCompletionDeadline = computeDeleteOperationCompletionDeadline(cluster)

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

	// recurse down to delete children
	nodePoolIterator, err := f.resourcesDBClient.HCPClusters(cluster.ID.SubscriptionID, cluster.ID.ResourceGroupName).NodePools(cluster.ID.Name).List(ctx, nil)
	if err != nil {
		return utils.TrackError(err)
	}
	for _, nodePool := range nodePoolIterator.Items(ctx) {
		// don't include the writer/request so that we don't have conflicting notificationURIs
		if err := f.addDeleteNodePoolToTransaction(ctx, nil, nil, transaction, nodePool); err != nil {
			return utils.TrackError(err)
		}
	}
	externalAuthIterator, err := f.resourcesDBClient.HCPClusters(cluster.ID.SubscriptionID, cluster.ID.ResourceGroupName).ExternalAuth(cluster.ID.Name).List(ctx, nil)
	if err != nil {
		return utils.TrackError(err)
	}
	for _, externalAuth := range externalAuthIterator.Items(ctx) {
		// don't include the writer/request so that we don't have conflicting notificationURIs
		if err := f.addDeleteExternalAuthToTransaction(ctx, nil, nil, transaction, externalAuth); err != nil {
			return utils.TrackError(err)
		}
	}

	return nil
}

func computeDeleteOperationCompletionDeadline(cluster *coreapi.HCPOpenShiftCluster) *metav1.Time {
	duration := admission.DefaultDeleteOperationCompletionDeadlineDuration
	if cluster.ServiceProviderProperties.DeleteOperationCompletionTimeout != nil {
		duration = *cluster.ServiceProviderProperties.DeleteOperationCompletionTimeout
	}
	deadline := metav1.NewTime(cluster.ServiceProviderProperties.DeletionTimestamp.Add(duration))
	return &deadline
}

func (f *Frontend) getInternalClusterFromStorage(ctx context.Context, resourceID *azcorearm.ResourceID) (*coreapi.HCPOpenShiftCluster, error) {
	internalCluster, err := f.resourcesDBClient.HCPClusters(resourceID.SubscriptionID, resourceID.ResourceGroupName).Get(ctx, resourceID.Name)
	if cosmosstorageutils.IsNotFoundError(err) {
		return nil, coreapi.NewResourceNotFoundError(resourceID)
	}
	if err != nil {
		return nil, utils.TrackError(err)
	}

	// Replace the ID field from Cosmos with the given resourceID,
	// which typically comes from the URL. This helps preserve the
	// casing of the resource group and resource name from the URL
	// to meet RPC requirements:
	//
	// Put Resource | Arguments
	//
	// The resource group names and resource names should be matched
	// case insensitively. ... Additionally, the Resource Provider must
	// preserve the casing provided by the user. The service must return
	// the most recently specified casing to the client and must not
	// normalize or return a toupper or tolower form of the resource
	// group or resource name. The resource group name and resource
	// name must come from the URL and not the request body.
	if internalCluster.ResourceID == nil {
		return nil, fmt.Errorf("stored cluster document is missing cosmosMetadata.resourceID")
	}
	if !strings.EqualFold(internalCluster.ResourceID.String(), resourceID.String()) {
		return nil, fmt.Errorf("unexpected resourceID: %s", internalCluster.ResourceID.String())
	}
	internalCluster.ID = resourceID

	// this allows partial information to be provided by a controller.
	completeClusterIdentity(internalCluster, nil)

	// temporarily fill in info if we have something missing.

	return internalCluster, nil
}

// ensureSystemData tries to use the src systemData
func ensureSystemData(newObj, oldObj *coreapi.SystemData) *coreapi.SystemData {
	var ret *coreapi.SystemData
	if newObj != nil {
		ret = newObj.DeepCopy()
	} else {
		ret = &coreapi.SystemData{}
	}
	if oldObj != nil {
		ret.CreatedAt = oldObj.CreatedAt
		ret.CreatedBy = oldObj.CreatedBy
		ret.CreatedByType = oldObj.CreatedByType
	}

	if ret.CreatedAt == nil || ret.CreatedAt.IsZero() {
		ret.CreatedAt = ptr.To(time.Now().UTC())
	}
	if len(ret.CreatedBy) == 0 {
		ret.CreatedBy = "Unknown-ARO-HCP-frontend"
	}
	if len(ret.CreatedByType) == 0 {
		ret.CreatedByType = coreapi.CreatedByTypeApplication
	}

	if ret.LastModifiedAt == nil || ret.LastModifiedAt.IsZero() {
		ret.LastModifiedAt = ptr.To(time.Now().UTC())
	}
	if len(ret.LastModifiedBy) == 0 {
		ret.LastModifiedBy = "Unknown-ARO-HCP-frontend"
	}
	if len(ret.LastModifiedByType) == 0 {
		ret.LastModifiedByType = coreapi.CreatedByTypeApplication
	}

	return ret
}

// completeClusterIdentity fills in any missing cluster.Identity.UserAssignedIdentities and removes any extra cluster.Identity.UserAssignedIdentities keys.
func completeClusterIdentity(cluster *coreapi.HCPOpenShiftCluster, existingUserAssignedIdentity map[string]*coreapi.UserAssignedIdentity) {
	allExpectedKeys := sets.Set[string]{}

	// set default .Identity.UserAssignedIdentities if none exist for required entry.
	for _, operatorIdentityResourceID := range cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ControlPlaneOperators {
		if operatorIdentityResourceID == nil {
			continue
		}
		allExpectedKeys.Insert(operatorIdentityResourceID.String())
		if cluster.Identity == nil {
			cluster.Identity = &coreapi.ManagedServiceIdentity{}
		}
		if cluster.Identity.UserAssignedIdentities == nil {
			cluster.Identity.UserAssignedIdentities = make(map[string]*coreapi.UserAssignedIdentity)
		}

		if val, ok := cluster.Identity.UserAssignedIdentities[operatorIdentityResourceID.String()]; !ok || val == nil {
			// existing entries can be present-but-nil (older Cosmos records that stored only
			// the keys), so a nil existingValue is treated as "no existing details" instead of
			// being copied through to a nil map entry that would serialize as `null`.
			if existingValue := existingUserAssignedIdentity[operatorIdentityResourceID.String()]; existingValue != nil {
				cluster.Identity.UserAssignedIdentities[operatorIdentityResourceID.String()] = existingValue.DeepCopy()
			} else {
				cluster.Identity.UserAssignedIdentities[operatorIdentityResourceID.String()] = &coreapi.UserAssignedIdentity{}
			}
		}
	}
	if serviceManagedIdentity := cluster.CustomerProperties.Platform.OperatorsAuthentication.UserAssignedIdentities.ServiceManagedIdentity; serviceManagedIdentity != nil {
		allExpectedKeys.Insert(serviceManagedIdentity.String())

		if cluster.Identity == nil {
			cluster.Identity = &coreapi.ManagedServiceIdentity{}
		}
		if cluster.Identity.UserAssignedIdentities == nil {
			cluster.Identity.UserAssignedIdentities = make(map[string]*coreapi.UserAssignedIdentity)
		}

		if val, ok := cluster.Identity.UserAssignedIdentities[serviceManagedIdentity.String()]; !ok || val == nil {
			// Same nil-existing handling as the control plane operators path above.
			if existingValue := existingUserAssignedIdentity[serviceManagedIdentity.String()]; existingValue != nil {
				cluster.Identity.UserAssignedIdentities[serviceManagedIdentity.String()] = existingValue.DeepCopy()
			} else {
				cluster.Identity.UserAssignedIdentities[serviceManagedIdentity.String()] = &coreapi.UserAssignedIdentity{}
			}
		}
	}

	if cluster.Identity != nil {
		for key := range cluster.Identity.UserAssignedIdentities {
			if !allExpectedKeys.Has(key) {
				delete(cluster.Identity.UserAssignedIdentities, key)
			}
		}
	}
}
