// 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/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"
)

func (f *Frontend) GetNodePool(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)
	}

	resultingInternalNodePool, err := f.getInternalNodePoolFromStorage(ctx, resourceID)
	if err != nil {
		return utils.TrackError(err)
	}
	responseBytes, err := coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftClusterNodePool(resultingInternalNodePool))
	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) ArmResourceListNodePools(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)
	clusterName := request.PathValue(PathSegmentResourceName)

	// Verify the parent cluster exists so we return 404 instead of an empty
	// list for a non-existent cluster (Cosmos List is prefix-based).
	if _, err := f.resourcesDBClient.HCPClusters(subscriptionID, resourceGroupName).Get(ctx, clusterName); err != nil {
		return utils.TrackError(err)
	}

	pagedResponse := coreapi.NewPagedResponse()

	internalNodePoolIterator, err := f.resourcesDBClient.HCPClusters(subscriptionID, resourceGroupName).NodePools(clusterName).List(ctx, dbListOptionsFromRequest(request))
	if err != nil {
		return utils.TrackError(err)
	}
	for _, nodePool := range internalNodePoolIterator.Items(ctx) {
		resultingExternalNodePool := versionedInterface.NewHCPOpenShiftClusterNodePool(nodePool)
		jsonBytes, err := coreapi.MarshalJSON(resultingExternalNodePool)
		if err != nil {
			return utils.TrackError(err)
		}
		pagedResponse.AddValue(jsonBytes)
	}
	err = internalNodePoolIterator.GetError()
	if err != nil {
		return utils.TrackError(err)
	}

	// MiddlewareReferer ensures Referer is present.
	err = pagedResponse.SetNextLink(request.Referer(), internalNodePoolIterator.GetContinuationToken())
	if err != nil {
		return utils.TrackError(err)
	}

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

func (f *Frontend) CreateOrUpdateNodePool(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()

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

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

	updating := oldInternalNodePool != nil
	if updating {
		if err := checkForProvisioningStateConflict(ctx, f.resourcesDBClient, cosmosstorageutils.OperationRequestUpdate, oldInternalNodePool.ID, oldInternalNodePool.Properties.ProvisioningState); err != nil {
			return utils.TrackError(err)
		}
		switch request.Method {
		case http.MethodPut:
			return f.updateNodePool(writer, request, oldInternalNodePool)
		case http.MethodPatch:
			return f.patchNodePool(writer, request, oldInternalNodePool)
		default:
			return fmt.Errorf("unsupported method %s", request.Method)
		}
	}

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

func decodeDesiredNodePoolCreate(ctx context.Context, azureLocation string) (*coreapi.HCPOpenShiftClusterNodePool, 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())
	}

	externalNodePoolFromRequest := versionedInterface.NewHCPOpenShiftClusterNodePool(nil)
	if err := json.Unmarshal(body, &externalNodePoolFromRequest); err != nil {
		return nil, utils.TrackError(err)
	}
	newInternalNodePool, err := externalNodePoolFromRequest.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.
	newInternalNodePool.EnsureDefaults()
	// TrackedResource info doesn't appear to come from the external resource information
	if len(newInternalNodePool.Name) > 0 && newInternalNodePool.Name != resourceID.Name {
		return nil, nameResourceIDMismatch(resourceID, newInternalNodePool.Name)
	}
	conversion.CopyReadOnlyTrackedResourceValues(&newInternalNodePool.TrackedResource, ptr.To(coreapi.NewTrackedResource(resourceID, azureLocation)))
	newInternalNodePool.SetResourceID(resourceID)
	newInternalNodePool.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.
	newInternalNodePool.SystemData = ensureSystemData(systemData, nil)

	return newInternalNodePool, nil
}

// newNodePoolAdmissionContext creates an admission context for node pool operations.
// The cluster parameter is always required.
// The spCluster and spNodePool parameters provide service provider state needed for
// admission checks that depend on runtime state (e.g., version upgrade validation).
// For UPDATE operations, these parameters are required and the function will fail if they're nil.
// For CREATE operations, these can be nil since no prior state exists.
func (f *Frontend) newNodePoolAdmissionContext(ctx context.Context, op operation.Operation, subscription *coreapi.Subscription, originalNodePool *coreapi.HCPOpenShiftClusterNodePool, cluster *coreapi.HCPOpenShiftCluster, spCluster *coreapi.ServiceProviderCluster, spNodePool *coreapi.ServiceProviderNodePool) (*admission.NodePoolAdmissionContext, error) {
	if cluster == nil {
		return nil, fmt.Errorf("cluster is required for admission context")
	}

	if op.Type == operation.Update {
		if spCluster == nil {
			return nil, fmt.Errorf("serviceProviderCluster is required for UPDATE operations")
		}
		if spNodePool == nil {
			return nil, fmt.Errorf("serviceProviderNodePool is required for UPDATE operations")
		}
	}

	return &admission.NodePoolAdmissionContext{
		Clock:                   f.clock,
		Subscription:            subscription,
		OriginalNodePool:        originalNodePool.DeepCopy(),
		Cluster:                 cluster,
		ServiceProviderCluster:  spCluster,
		ServiceProviderNodePool: spNodePool,
	}, nil
}

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

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

	newInternalNodePool, err := decodeDesiredNodePoolCreate(ctx, f.azureLocation)
	if err != nil {
		return utils.TrackError(err)
	}

	subscription, err := f.resourcesDBClient.Subscriptions().Get(ctx, resourceID.SubscriptionID)
	if err != nil {
		return err
	}

	// Node pool validation checks some fields against the parent cluster
	// so we have to request the cluster from Cluster Service.
	cluster, err := f.getInternalClusterFromStorage(ctx, resourceID.Parent)
	if err != nil {
		return utils.TrackError(err)
	}
	if cluster.ServiceProviderProperties.ClusterServiceID == nil {
		return utils.TrackError(fmt.Errorf("cluster %s has no ClusterServiceID", cluster.ID))
	}

	restOperation := operation.Operation{
		Type:    operation.Create,
		Options: validation.BuildValidationOptions(subscription.GetRegisteredFeatures(), metadataapi.APIVersion(versionedInterface.String())),
	}
	admissionContext, err := f.newNodePoolAdmissionContext(ctx, restOperation, subscription, newInternalNodePool, cluster, nil, nil)
	if err != nil {
		return utils.TrackError(err)
	}
	if mutationErrs := admission.MutateNodePool(ctx, admissionContext, restOperation, newInternalNodePool, nil); len(mutationErrs) > 0 {
		return utils.TrackError(coreapi.CloudErrorFromFieldErrors(mutationErrs))
	}

	validationErrs := validation.ValidateNodePool(ctx, restOperation, newInternalNodePool, nil)
	// in addition to static validation, we have validation based on the state of the hcp cluster
	validationErrs = append(validationErrs, admission.AdmitNodePool(ctx, admissionContext, restOperation, newInternalNodePool, nil)...)
	if err := coreapi.CloudErrorFromFieldErrors(validationErrs); err != nil {
		return utils.TrackError(err)
	}

	logger.Info(fmt.Sprintf("creating resource %s", resourceID))
	if err := checkForProvisioningStateConflict(ctx, f.resourcesDBClient, cosmosstorageutils.OperationRequestCreate, newInternalNodePool.ID, newInternalNodePool.Properties.ProvisioningState); err != nil {
		return utils.TrackError(err)
	}

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

	createNodePoolOperation := cosmosstorageutils.NewOperation(
		cosmosstorageutils.OperationRequestCreate,
		newInternalNodePool.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, createNodePoolOperation.NotificationURI, createNodePoolOperation.OperationID))
	_, err = f.resourcesDBClient.Operations(newInternalNodePool.ID.SubscriptionID).AddCreateToTransaction(ctx, transaction, createNodePoolOperation, 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.
	newInternalNodePool.ServiceProviderProperties.ActiveOperationID = createNodePoolOperation.ResourceID.Name
	newInternalNodePool.Properties.ProvisioningState = createNodePoolOperation.Status

	nodePoolCosmosClient := f.resourcesDBClient.HCPClusters(resourceID.SubscriptionID, resourceID.ResourceGroupName).NodePools(resourceID.Parent.Name)
	cosmosUID, err := nodePoolCosmosClient.AddCreateToTransaction(ctx, transaction, newInternalNodePool, 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.
	resultingUncastInternalNodePool, err := transactionResult.GetItem(cosmosUID)
	if err != nil {
		return utils.TrackError(err)
	}
	resultingInternalNodePool, ok := resultingUncastInternalNodePool.(*coreapi.HCPOpenShiftClusterNodePool)
	if !ok {
		return fmt.Errorf("unexpected type %T", resultingUncastInternalNodePool)
	}
	responseBytes, err := coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftClusterNodePool(resultingInternalNodePool))
	if err != nil {
		return utils.TrackError(err)
	}

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

func decodeDesiredNodePoolReplace(ctx context.Context, oldInternalNodePool *coreapi.HCPOpenShiftClusterNodePool) (*coreapi.HCPOpenShiftClusterNodePool, 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)
	}

	// Initialize versionedRequestNodePool to include both
	// non-zero default values and current read-only values.
	// Exact user request
	externalNodePoolFromRequest := versionedInterface.NewHCPOpenShiftClusterNodePool(nil)
	if err := json.Unmarshal(body, &externalNodePoolFromRequest); err != nil {
		return nil, utils.TrackError(err)
	}

	newInternalNodePool, err := externalNodePoolFromRequest.ConvertToInternal(oldInternalNodePool)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	if len(newInternalNodePool.Name) > 0 && newInternalNodePool.Name != resourceID.Name {
		return nil, nameResourceIDMismatch(resourceID, newInternalNodePool.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(newInternalNodePool.Properties.Version.ID) == 0 {
		newInternalNodePool.Properties.Version.ID = oldInternalNodePool.Properties.Version.ID
	}
	if newInternalNodePool.Properties.Platform.SubnetID == nil {
		newInternalNodePool.Properties.Platform.SubnetID = oldInternalNodePool.Properties.Platform.SubnetID
	}

	// 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.CopyReadOnlyNodePoolValues(newInternalNodePool, oldInternalNodePool)
	newInternalNodePool.SystemData = ensureSystemData(systemData, oldInternalNodePool.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 newInternalNodePool.Tags == nil {
		newInternalNodePool.Tags = maps.Clone(oldInternalNodePool.Tags)
	}

	// Clear the user-assigned identities map since that is reconstructed from Cluster Service data.
	// TODO we'd like to have the instance complete when we go to validate it.  Right now validation fails if we clear this.
	// TODO we probably update validation to require this field is cleared.
	//newInternalCluster.Identity.UserAssignedIdentities = nil

	// Backstop for fields unknown to this API version's SetDefaultValues*.
	// See docs/api-version-defaults-and-storage.md.
	newInternalNodePool.EnsureDefaults()

	return newInternalNodePool, nil
}

func (f *Frontend) updateNodePool(writer http.ResponseWriter, request *http.Request, oldInternalNodePool *coreapi.HCPOpenShiftClusterNodePool) error {
	ctx := request.Context()

	newInternalNodePool, err := decodeDesiredNodePoolReplace(ctx, oldInternalNodePool)
	if err != nil {
		return utils.TrackError(err)
	}

	return f.updateNodePoolInCosmos(ctx, writer, request, http.StatusOK, newInternalNodePool, oldInternalNodePool)
}

func decodeDesiredNodePoolPatch(ctx context.Context, oldInternalNodePool *coreapi.HCPOpenShiftClusterNodePool) (*coreapi.HCPOpenShiftClusterNodePool, 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.
	newExternalNodePool := versionedInterface.NewHCPOpenShiftClusterNodePool(oldInternalNodePool)
	if err := coreapi.ApplyRequestBody(http.MethodPatch, body, newExternalNodePool); err != nil {
		return nil, utils.TrackError(err)
	}
	newInternalNodePool, err := newExternalNodePool.ConvertToInternal(oldInternalNodePool)
	if err != nil {
		return nil, utils.TrackError(err)
	}
	if len(newInternalNodePool.Name) > 0 && newInternalNodePool.Name != resourceID.Name {
		return nil, nameResourceIDMismatch(resourceID, newInternalNodePool.Name)
	}

	conversion.CopyReadOnlyNodePoolValues(newInternalNodePool, oldInternalNodePool)
	newInternalNodePool.SystemData = ensureSystemData(systemData, oldInternalNodePool.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 newInternalNodePool.Tags == nil {
		newInternalNodePool.Tags = maps.Clone(oldInternalNodePool.Tags)
	}

	// Clear the user-assigned identities map since that is reconstructed from Cluster Service data.
	// TODO we'd like to have the instance complete when we go to validate it.  Right now validation fails if we clear this.
	// TODO we probably update validation to require this field is cleared.
	//newInternalCluster.Identity.UserAssignedIdentities = nil

	// Backstop for fields unknown to this API version's SetDefaultValues*.
	// See docs/api-version-defaults-and-storage.md.
	newInternalNodePool.EnsureDefaults()

	return newInternalNodePool, nil
}

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

	newInternalNodePool, err := decodeDesiredNodePoolPatch(ctx, oldInternalNodePool)
	if err != nil {
		return utils.TrackError(err)
	}

	return f.updateNodePoolInCosmos(ctx, writer, request, http.StatusAccepted, newInternalNodePool, oldInternalNodePool)
}

func (f *Frontend) updateNodePoolInCosmos(ctx context.Context, writer http.ResponseWriter, request *http.Request, httpStatusCode int, newInternalNodePool, oldInternalNodePool *coreapi.HCPOpenShiftClusterNodePool) error {
	logger := utils.LoggerFromContext(ctx)

	subscription, err := f.resourcesDBClient.Subscriptions().Get(ctx, oldInternalNodePool.ID.SubscriptionID)
	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)
	}
	// Node pool validation checks some fields against the parent cluster
	// so we have to request the cluster from Cluster Service.
	cluster, err := f.getInternalClusterFromStorage(ctx, oldInternalNodePool.ID.Parent)
	if err != nil {
		return utils.TrackError(err)
	}

	// Get ServiceProviderCluster and ServiceProviderNodePool for version validation
	clusterID := oldInternalNodePool.ID.Parent
	spCluster, err := corecosmosstorage.GetOrCreateServiceProviderCluster(ctx, f.resourcesDBClient, clusterID)
	if err != nil {
		return utils.TrackError(err)
	}

	spNodePool, err := corecosmosstorage.GetOrCreateServiceProviderNodePool(ctx, f.resourcesDBClient, oldInternalNodePool.ID)
	if err != nil {
		return utils.TrackError(err)
	}

	restOperation := operation.Operation{
		Type:    operation.Update,
		Options: validation.BuildValidationOptions(subscription.GetRegisteredFeatures(), metadataapi.APIVersion(versionedInterface.String())),
	}
	admissionContext, err := f.newNodePoolAdmissionContext(ctx, restOperation, subscription, newInternalNodePool, cluster, spCluster, spNodePool)
	if err != nil {
		return utils.TrackError(err)
	}
	if mutationErrs := admission.MutateNodePool(ctx, admissionContext, restOperation, newInternalNodePool, oldInternalNodePool); len(mutationErrs) > 0 {
		return utils.TrackError(coreapi.CloudErrorFromFieldErrors(mutationErrs))
	}

	validationErrs := validation.ValidateNodePool(ctx, restOperation, newInternalNodePool, oldInternalNodePool)
	validationErrs = append(validationErrs, admission.AdmitNodePool(ctx, admissionContext, restOperation, newInternalNodePool, oldInternalNodePool)...)
	if err := coreapi.CloudErrorFromFieldErrors(validationErrs); err != nil {
		return utils.TrackError(err)
	}

	logger.Info(fmt.Sprintf("updating resource %s", oldInternalNodePool.ID))

	// The cosmos representation the new desired version
	// The controllers will take care of handle the upgrade

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

	nodePoolUpdateOperation := cosmosstorageutils.NewOperation(
		cosmosstorageutils.OperationRequestUpdate,
		newInternalNodePool.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, nodePoolUpdateOperation.NotificationURI, nodePoolUpdateOperation.OperationID))
	_, err = f.resourcesDBClient.Operations(newInternalNodePool.ID.SubscriptionID).AddCreateToTransaction(ctx, transaction, nodePoolUpdateOperation, 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.
	newInternalNodePool.ServiceProviderProperties.ActiveOperationID = nodePoolUpdateOperation.ResourceID.Name
	newInternalNodePool.Properties.ProvisioningState = nodePoolUpdateOperation.Status

	_, err = f.resourcesDBClient.HCPClusters(newInternalNodePool.ID.SubscriptionID, newInternalNodePool.ID.ResourceGroupName).
		NodePools(newInternalNodePool.ID.Parent.Name).
		AddReplaceToTransaction(ctx, transaction, newInternalNodePool, 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.
	resultingUncastInternalNodePool, err := transactionResult.GetItem(oldInternalNodePool.GetCosmosData().GetCosmosUID())
	if err != nil {
		return utils.TrackError(err)
	}
	resultingInternalNodePool, ok := resultingUncastInternalNodePool.(*coreapi.HCPOpenShiftClusterNodePool)
	if !ok {
		return fmt.Errorf("unexpected type %T", resultingUncastInternalNodePool)
	}
	responseBytes, err := coreapi.MarshalJSON(versionedInterface.NewHCPOpenShiftClusterNodePool(resultingInternalNodePool))
	if err != nil {
		return utils.TrackError(err)
	}

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

// DeleteNodePool 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) DeleteNodePool(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)
	}

	nodePool, err := f.resourcesDBClient.HCPClusters(resourceID.SubscriptionID, resourceID.ResourceGroupName).NodePools(resourceID.Parent.Name).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, nodePool.ID, nodePool.Properties.ProvisioningState); err != nil {
		return utils.TrackError(err)
	}

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

	// We retrieve all node pools for the cluster, including the one we are attempting to
	// delete, and we pass them to the delete admission validation.
	// TODO once OCPBUGS-86702 is resolved, we should remove this retrieval and the check of last nodepool being
	// deleted in the delete admission validation when we decide we want to allow the deletion of the last node pool.
	nodePoolIterator, err := f.resourcesDBClient.HCPClusters(nodePool.ID.SubscriptionID, nodePool.ID.ResourceGroupName).NodePools(nodePool.ID.Parent.Name).List(ctx, nil)
	if err != nil {
		return utils.TrackError(err)
	}
	clusterNodePools := make([]*coreapi.HCPOpenShiftClusterNodePool, 0)
	for _, clusterNodePool := range nodePoolIterator.Items(ctx) {
		clusterNodePools = append(clusterNodePools, clusterNodePool)
	}
	if err := nodePoolIterator.GetError(); err != nil {
		return utils.TrackError(err)
	}

	nodePoolDeleteAdmissionContext := &admission.NodePoolDeleteAdmissionContext{
		ClusterNodePools: clusterNodePools,
	}
	err = coreapi.CloudErrorFromFieldErrors(admission.AdmitNodePoolOnDelete(ctx, nodePoolDeleteAdmissionContext, nodePool))
	if err != nil {
		return utils.TrackError(err)
	}

	transaction := f.resourcesDBClient.NewTransaction(nodePool.ID.SubscriptionID)
	if err := f.addDeleteNodePoolToTransaction(ctx, writer, request, transaction, nodePool); 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) addDeleteNodePoolToTransaction(ctx context.Context, writer http.ResponseWriter, request *http.Request, transaction cosmosstorageutils.DBTransaction, nodePool *coreapi.HCPOpenShiftClusterNodePool) 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:             nodePool.ID,
		IncludeNestedResources: true,
	})
	if err != nil {
		return utils.TrackError(err)
	}

	operationDoc := cosmosstorageutils.NewOperation(
		cosmosstorageutils.OperationRequestDelete,
		nodePool.ID,
		metadataapi.InternalID{},
		f.azureLocation,
		"",
		"",
		"",
		correlationData)
	// TODO remove this once migration of the new node pool deletion from frontend to backend approach is fully completed in all ARO-HCP
	// permanent environments, for all regions.
	operationDoc.UsesNewNodePoolDeletionApproach = 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 nodePool.ServiceProviderProperties.DeletionTimestamp == nil {
		nodePool.ServiceProviderProperties.DeletionTimestamp = &metav1.Time{Time: time.Now().UTC()}
	}
	nodePool.ServiceProviderProperties.ActiveOperationID = operationDoc.ResourceID.Name
	nodePool.Properties.ProvisioningState = operationDoc.Status
	// TODO remove this once migration of the new node pool deletion from frontend to backend approach is fully completed in all ARO-HCP
	// permanent environments, for all regions.
	nodePool.ServiceProviderProperties.UsesNewNodePoolDeletionApproach = true
	_, err = f.resourcesDBClient.HCPClusters(nodePool.ID.SubscriptionID, nodePool.ID.ResourceGroupName).NodePools(nodePool.ID.Parent.Name).
		AddReplaceToTransaction(ctx, transaction, nodePool, nil)
	if err != nil {
		return utils.TrackError(err)
	}

	return nil
}

func (f *Frontend) getInternalNodePoolFromStorage(ctx context.Context, resourceID *azcorearm.ResourceID) (*coreapi.HCPOpenShiftClusterNodePool, error) {
	internalNodePool, err := f.resourcesDBClient.HCPClusters(resourceID.SubscriptionID, resourceID.ResourceGroupName).NodePools(resourceID.Parent.Name).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 internalNodePool.ResourceID == nil {
		return nil, fmt.Errorf("stored nodepool document is missing cosmosMetadata.resourceID")
	}
	if !strings.EqualFold(internalNodePool.ResourceID.String(), resourceID.String()) {
		return nil, fmt.Errorf("unexpected resourceID: %s", internalNodePool.ResourceID.String())
	}
	internalNodePool.ID = resourceID

	return internalNodePool, nil

}
