package cluster

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cvt"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (svc *clusterServiceImpl) Update(
	ctx context.Context,
	req *globalserverlessv1.ServerlessServiceUpdateRequest,
) (*globalserverlessv1.ServerlessServiceUpdateResponse, error) {
	if req.Cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty request body")
	}

	ctx, span := Tracer.Start(ctx, "Domain.UpdateCluster")
	defer span.End()

	// Get cluster from repo.
	cluster, err := svc.clusterRepo.Get(ctx, req.Cluster.ClusterId, false)
	if err != nil {
		return nil, fmt.Errorf("load cluster failed: %w", err)
	}

	if !isPGStarterCluster(cluster) {
		// Get cluster from regional.
		regionalClusters, err := svc.regionalService.ListCluster(
			ctx,
			cluster.Region,
			&globalserverlessv1.ServerlessServiceListRequest{
				Filter: &globalserverlessv1.ClusterFilter{
					ClusterIds: []string{cluster.ClusterId},
					ProjectIds: []uint64{cluster.ProjectId},
				},
			},
		)
		if err != nil {
			return nil, fmt.Errorf("get regional cluster failed: %w", err)
		}
		if len(regionalClusters) != 1 {
			return nil, fmt.Errorf("cluster %s not found in regional", cluster.ClusterId)
		}
		cluster.Cluster = regionalClusters[0]
	}

	// Build output cluster.
	out := cvt.ClusterEnToGlobal(cluster)
	if err := svc.appendClusterDetailField(ctx, out, nil); err != nil {
		tilog.Error(ctx, "append cluster detail fields failed", zap.String("id", cluster.ClusterId), zap.Error(err))
	}

	// update cluster spend limit
	if req.GetCluster().GetSpendLimit() != nil {
		// disable essential plan downgrade
		if cluster.GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
			cluster.GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "essential plan can not downgrade")
		}
		// check can create scalable cluster
		canCreate, err := svc.billingFacade.CanCreateScalableCluster(
			ctx,
			cluster.Region.Id.String(),
			cluster.Region.CloudProviderId.String(),
			adapter.BillingResourceTypeServerlessCluster,
			cluster.TenantId,
		)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "check tenant plan failed: %v", err)
		}
		if !canCreate.Pass {
			return nil, bizerr.New(bizerr.ErrCodeNoPaymentMethods, "%s", canCreate.Reason)
		}
		spendingLimit, err := svc.serverlessFacade.GetSpendLimit(
			ctx,
			cluster.ClusterId,
			cluster.TenantId,
		)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "get spend limit measure detail failed: %w", err)
		}
		if spendingLimit.GetMonthly() > 0 && req.GetCluster().GetSpendLimit().GetMonthly() == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "scalable cluster can not set spending limit to 0")
		}
		if err = svc.serverlessFacade.SetSpendLimit(ctx,
			cluster.ClusterId,
			cluster.TenantId,
			req.GetCluster().GetSpendLimit().GetMonthly(),
			cluster.Region.Id.String(),
			false,
			true); err != nil {
			return nil, fmt.Errorf("update spendLimit check failed: %w", err)
		}

		cluster.SpendLimit = req.GetCluster().GetSpendLimit()
	}
	// update auto scaling
	if req.GetCluster().GetCluster().GetAutoScaling() != nil {
		canCreate, err := svc.billingFacade.CanCreateScalableCluster(
			ctx,
			cluster.Region.Id.String(),
			cluster.Region.CloudProviderId.String(),
			adapter.BillingResourceTypeEssentialCluster,
			cluster.TenantId,
		)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "check tenant plan failed: %v", err)
		}
		if !canCreate.Pass {
			return nil, bizerr.New(bizerr.ErrCodeNoPaymentMethods, "%s", canCreate.Reason)
		}
		if cluster.Cluster.AutoScaling == nil {
			cluster.Cluster.AutoScaling = &regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AutoScaling{}
		}
		if req.GetCluster().GetCluster().GetAutoScaling().GetMinRcu() > 0 {
			cluster.Cluster.AutoScaling.MinRcu = req.GetCluster().GetCluster().GetAutoScaling().GetMinRcu()
		}
		if req.GetCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
			cluster.Cluster.AutoScaling.MaxRcu = req.GetCluster().GetCluster().GetAutoScaling().GetMaxRcu()
		}
	}

	if req.GetCluster().GetBackupPolicy() != nil {
		cluster.BackupPolicy = req.GetCluster().GetBackupPolicy()
	}

	if publicEndpoint := req.GetCluster().GetCluster().GetEndpoints().GetPublic(); publicEndpoint != nil {
		if cluster.Cluster.Endpoints == nil {
			cluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
		}
		if cluster.Cluster.Endpoints.Public == nil {
			cluster.Cluster.Endpoints.Public = &regionalserverlessv1.PublicEndpoint{}
		}
		cluster.Cluster.Endpoints.Public.Disabled = publicEndpoint.GetDisabled()
		cluster.Cluster.Endpoints.Public.AuthorizedNetwork = publicEndpoint.GetAuthorizedNetwork()
	}

	if privateEndpoint := req.GetCluster().GetCluster().GetEndpoints().GetPrivate(); privateEndpoint != nil {
		if cluster.Cluster.Endpoints == nil {
			cluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
		}
		if cluster.Cluster.Endpoints.Private == nil {
			cluster.Cluster.Endpoints.Private = &regionalserverlessv1.PrivateEndpoint{}
		}
		cluster.Cluster.Endpoints.Private.AuthorizedNetwork = privateEndpoint.GetAuthorizedNetwork()
	}

	if req.GetCluster().GetCluster().GetAuditLogConfig() != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, constant.AuditLogDeprecatedMsg)
	}

	// update ru_version if set (> 0)
	if req.GetCluster().GetRuVersion() > 0 {
		cluster.RuVersion = req.GetCluster().GetRuVersion()
	}

	// Early return if validate only.
	if req.GetValidateOnly() {
		return &globalserverlessv1.ServerlessServiceUpdateResponse{
			Cluster: out,
		}, nil
	}

	// generate cluster event
	var updateEvents []clusterevent.Event
	publicEndpoint := req.GetCluster().GetCluster().GetEndpoints().GetPublic()
	if publicEndpoint != nil {
		eventType := clusterevent.EventTypeUpdatingPublicEndpoint
		ev, err := entity.NewClusterEvent(ctx, cluster, eventType, cluster.GetUserStatus().String())
		if err != nil {
			tilog.Error(
				ctx,
				"create cluster event failed",
				zap.String("id", cluster.ClusterId),
				zap.Any("eventtype", eventType),
				zap.Error(err),
			)
			return nil, err
		}
		updateEvents = append(updateEvents, ev)
	}
	if req.GetCluster().GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
		req.GetCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		eventType := clusterevent.EventTypeUpdatingCapacity
		ev, err := entity.NewClusterEvent(ctx, cluster, eventType, cluster.GetUserStatus().String())
		if err != nil {
			tilog.Error(
				ctx,
				"create cluster event failed",
				zap.String("id", cluster.ClusterId),
				zap.Any("eventtype", eventType),
				zap.Error(err),
			)
			return nil, err
		}
		updateEvents = append(updateEvents, ev)
	}
	if req.GetCluster().GetCluster().GetAuditLogConfig() != nil {
		eventType := clusterevent.EventTypeUpdatingAuditLogConfig
		ev, err := entity.NewClusterEvent(ctx, cluster, eventType, cluster.GetUserStatus().String())
		if err != nil {
			tilog.Error(
				ctx,
				"create cluster event failed",
				zap.String("id", cluster.ClusterId),
				zap.Any("eventtype", eventType),
				zap.Error(err),
			)
			return nil, err
		}
		alc, _ := json.Marshal(req.GetCluster().GetCluster().GetAuditLogConfig())
		ev.ExtraInfo.AuditLogConfig = string(alc)
		updateEvents = append(updateEvents, ev)
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Update ru_version in sgs DB if changed.
		if req.GetCluster().GetRuVersion() > 0 {
			if err := svc.clusterRepo.UpdateRuVersion(ctx, cluster.ClusterId, cluster.RuVersion); err != nil {
				return fmt.Errorf("update ru_version in db failed: %w", err)
			}
		}

		// Commit cluster event.
		if err := svc.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, updateEvents); err != nil {
			return err
		}

		requestStr, err := protojson.Marshal(req)
		if err != nil {
			return fmt.Errorf("marshal request failed: %w", err)
		}

		// Run workflow.
		if err := svc.flow.RunDagWithBizId(
			ctx,
			common.UpdateCluster,
			fmt.Sprintf("%d", cluster.TenantId),
			fmt.Sprintf("%d", cluster.ProjectId),
			cluster.ClusterId,
			map[string]string{
				workflow.KeyTraceId:   workflow.GetTraceId(ctx),
				common.KeyClusterId:   cluster.ClusterId,
				common.KeyRequestBody: string(requestStr),
			},
		); err != nil {
			return err
		}

		return nil
	}); err != nil {
		return nil, fmt.Errorf("update cluster failed: %w", err)
	}

	return &globalserverlessv1.ServerlessServiceUpdateResponse{
		Cluster: out,
	}, nil
}