package applier

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	eventent "github.com/tidbcloud/devops/entity/event"
	"github.com/tidbcloud/devops/pkg/infra/clients/configcentercli"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/clients/mgmthttpcli"
	"github.com/tidbcloud/devops/pkg/infra/eventbus"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/infra/utils/pointer"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"
	evbus "github.com/tidbcloud/eventbus-datastruct/event"

	clusterng "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"google.golang.org/protobuf/encoding/protojson"
)

type ResourcePoolApplierRequest struct {
	ClusterType ccproto.ClusterType
	ClusterId   string

	SharedPoolID  string
	LogicalRegion string
	CloudProvider string

	CCRequest     *ccproto.UpdateClusterMetaRequest
	ClusterSvcReq *clusterng.SetSharedPoolComponentConfigVersionRequest
}

func (r *ResourcePoolApplierRequest) ModifyConfigurationFile(file string, configType string,
	set func(ccreq *ccproto.UpdateClusterConfigurationFileRequest) error,
) error {

	var ccreq *ccproto.UpdateClusterConfigurationFileRequest

	for _, tmp := range r.CCRequest.ConfigurationFileRequests {
		if tmp.ConfigurationFilename == file {
			ccreq = tmp
			break
		}
	}

	if ccreq == nil {
		ccreq = &ccproto.UpdateClusterConfigurationFileRequest{
			ConfigurationFilename: file,
			ConfigurationType:     &configType,
		}
		r.CCRequest.ConfigurationFileRequests = append(r.CCRequest.ConfigurationFileRequests, ccreq)
	}

	return set(ccreq)
}

func (r *ResourcePoolApplierRequest) ModifyComponent(comp string,
	set func(ccReq *ccproto.UpdateClusterComponentRequest) error,
) error {

	var ccreq *ccproto.UpdateClusterComponentRequest

	for _, tmpReq := range r.CCRequest.ComponentRequests {
		if tmpReq.ComponentName == comp {
			ccreq = tmpReq
			break
		}
	}

	if ccreq == nil {
		ccreq = &ccproto.UpdateClusterComponentRequest{
			ComponentName: comp,
		}
		r.CCRequest.ComponentRequests = append(r.CCRequest.ComponentRequests, ccreq)
	}

	return set(ccreq)
}

type ResourcePoolApplier struct {
	coresvc      *coresvccli.CoreSvcCli
	configcenter *configcentercli.Client
	mgmthttpcli  mgmthttpcli.MgmtHTTPClient
	eventBus     *eventbus.EventBus
}

func NewResourcePoolApplier(coresvc *coresvccli.CoreSvcCli, mgmthttpcli mgmthttpcli.MgmtHTTPClient, configcenter *configcentercli.Client, eventbus *eventbus.EventBus) *ResourcePoolApplier {
	return &ResourcePoolApplier{
		coresvc:      coresvc,
		mgmthttpcli:  mgmthttpcli,
		configcenter: configcenter,
		eventBus:     eventbus,
	}
}

func (a *ResourcePoolApplier) Name() string {
	return "ResourcePoolApplier"
}

func (a *ResourcePoolApplier) AcceptVersionChange(verChange *ccent.ConfigurationVersionChange, changeType string) bool {
	return verChange != nil && changeType == string(api.ChangeTypeSharedPool)
}

func (a *ResourcePoolApplier) AcceptComponentChange(compChg *ccent.ConfigurationComponentChange) bool {
	return a.acceptComponent(compChg.Component)
}

func (a *ResourcePoolApplier) AcceptFileChange(fileChg *ccent.ConfigurationFileChange) bool {
	return a.acceptFile(fileChg.ConfigType)
}

func (a *ResourcePoolApplier) acceptComponent(comp string) bool {
	return slices.Contains(ccproto.ClusterType_SHARED_POOL.AvailableComponents(), comp)
}

func (a *ResourcePoolApplier) acceptFile(configType string) bool {
	return slices.Contains(ccproto.ClusterType_SHARED_POOL.AvailableConfigTypes(), configType)
}

func (a *ResourcePoolApplier) Prepare(ctx context.Context, actx *ApplyContext) error {
	chg := actx.Input.Change

	clusterIDInt, err := strconv.ParseUint(chg.TargetCluster.ClusterID, 10, 64)
	if err != nil {
		return fmt.Errorf("parse cluster id to int failed: %w", err)
	}
	spID := chg.TargetCluster.ClusterID

	sp, err := a.coresvc.Mgmt.GetResourcePool(ctx, clusterIDInt)
	if err != nil {
		// return fmt.Errorf("get resource pool failed: %w", err)
		// legacy ui: use premium cluster as "cluster id" in ui&api, convert to shared pool id in devops
		cluster, err := a.coresvc.Mgmt.GetCluster(ctx, clusterIDInt)
		if err == nil {
			spID = cluster.GetNextGenCluster().GetSharedPoolId()
			sp, err = a.coresvc.Mgmt.GetResourcePoolString(ctx, spID)
			if err != nil {
				return fmt.Errorf("get resource pool failed: %w", err)
			}
		}
	}

	actx.ResourcePoolApplierRequest = &ResourcePoolApplierRequest{
		ClusterType:   cvt.To.ConfigCenter.ClusterType(chg.TargetCluster.ClusterType),
		ClusterId:     chg.TargetCluster.ClusterID,
		SharedPoolID:  spID,
		LogicalRegion: sp.GetRegion().GetLogicalName(),
		CloudProvider: sp.GetRegion().GetCloudProviderId().String(),

		ClusterSvcReq: &clusterng.SetSharedPoolComponentConfigVersionRequest{
			SharedPoolId: spID,
			Components:   []*clusterng.TargetComponent{}, // set later
			//Scope: ,
			//Mode:, set later
			Force: chg.ApplyPolicy == ccent.ParameterApplyPolicyForceRestart,
		},
		CCRequest: &ccproto.UpdateClusterMetaRequest{
			ClusterType:               cvt.To.ConfigCenter.ClusterType(chg.TargetCluster.ClusterType),
			ClusterId:                 spID,
			Version:                   nil, // set later
			ComponentRequests:         nil, // set later
			ConfigurationFileRequests: nil, // set later
		},
	}
	request := actx.ResourcePoolApplierRequest
	var sharedPool *mgmthttpcli.SharedPool

	if actx.Input.Version != nil {
		request.CCRequest.Version = pointer.Pointer(actx.Input.Version.To)
	}

	getSharedPool := func() (*mgmthttpcli.SharedPool, error) {
		if sharedPool != nil {
			return sharedPool, nil
		}
		sp, err := a.mgmthttpcli.NextGenRegionServer(request.LogicalRegion, request.CloudProvider).GetSharedPool(ctx, request.SharedPoolID)
		if err != nil {
			return nil, fmt.Errorf("get shared pool failed: %w", err)
		}
		sharedPool = sp
		return sharedPool, nil
	}

	// update configuration files from file changes.
	for _, fileChg := range actx.Input.Files {
		filename := fileChg.Filename
		configType := fileChg.ConfigType

		err := request.ModifyConfigurationFile(filename, configType,
			func(ccreq *ccproto.UpdateClusterConfigurationFileRequest) error {
				//  build the request for the config center
				if fileChg.ModifiedTemplate != nil {
					ccreq.TemplateId = pointer.Pointer(int64(fileChg.ModifiedTemplate.To))
				}
				ccreq.CustomParams = map[string]*ccproto.ParamItem{}
				for _, item := range fileChg.AllParameters {
					if item.Source != ccent.ParameterCustom { // skip non-custom parameters
						continue
					}

					param, err := cvt.To.ConfigCenter.NamedParameter(item.NamedParameter)
					if err != nil {
						return fmt.Errorf("convert cluster config item failed: %w", err)
					}

					ccreq.CustomParams[item.Name] = param
				}
				ccreq.PreviousText = pointer.Pointer(fileChg.PreviousText)
				ccreq.AllowMissing = false // only modify, never create
				ccreq.ClearCustomParams = fileChg.ClearCustomParams

				return nil
			},
		)
		if err != nil {
			return err
		}
	}

	for _, compChg := range actx.Input.Components {
		err := request.ModifyComponent(compChg.Component,
			func(ccReq *ccproto.UpdateClusterComponentRequest) error {
				if compChg.ModifiedVersion != nil {
					ccReq.Version = &compChg.ModifiedVersion.To
				} else if actx.Input.Version != nil {
					ccReq.Version = &actx.Input.Version.To
				}
				if compChg.ModifiedOverwriteImageTag != nil {
					ccReq.OverwriteImageTag = &compChg.ModifiedOverwriteImageTag.To
				}

				return nil
			},
		)
		if err != nil {
			return err
		}
	}

	err = a.setResourcePoolTargetComponents(ctx, request, getSharedPool)
	if err != nil {
		return err
	}

	if chg.ApplyPolicy == ccent.ParameterApplyPolicyHotReload {
		request.ClusterSvcReq.Mode = clusterng.SetSharedPoolComponentConfigVersionRequest_HOT_RELOAD
	} else {
		request.ClusterSvcReq.Mode = clusterng.SetSharedPoolComponentConfigVersionRequest_ROLLING_UPDATE
	}
	if len(request.CCRequest.ComponentRequests) > 0 {
		if len(request.CCRequest.ConfigurationFileRequests) > 0 {
			request.ClusterSvcReq.Scope = clusterng.SetSharedPoolComponentConfigVersionRequest_ALL
		} else {
			request.ClusterSvcReq.Scope = clusterng.SetSharedPoolComponentConfigVersionRequest_VERSION
		}
	} else {
		if len(request.CCRequest.ConfigurationFileRequests) > 0 {
			request.ClusterSvcReq.Scope = clusterng.SetSharedPoolComponentConfigVersionRequest_CONFIG
		} else {
			return fmt.Errorf("no changes to apply for resource pool")
		}
	}

	// check changeable
	if chg.ApplyPolicy != ccent.ParameterApplyPolicyForceRestart {
		sharedPool, err := getSharedPool()
		if err != nil {
			return err
		}
		targets := componentGroupAvailabilityTargets(request.CCRequest.GetClusterType(), request.ClusterSvcReq.Components)
		if len(targets) > 0 {
			done, reason := targetComponentsAvailable(sharedPool.ComponentGroups, targets)
			if !done {
				return fmt.Errorf("%s, not changeable", reason)
			}
		}
	}

	return nil
}

func (a *ResourcePoolApplier) setResourcePoolTargetComponents(ctx context.Context, request *ResourcePoolApplierRequest, getSharedPool func() (*mgmthttpcli.SharedPool, error)) error {
	// Shared-pool TiKV config changes target TiKV groups by config-center
	// filenames: tikv-<tikv_group_id>. The legacy bare "tikv" file maps to the
	// shared-pool-level TiKV target without a group id.
	updatedTikvID := make(map[string]bool)
	updatedComponent := make(map[string]bool)
	for _, cfgreq := range request.CCRequest.ConfigurationFileRequests {
		if cfgreq.GetConfigurationType() == ccproto.SHARED_POOL_CONFIG_TIKV {
			updatedTikvID[filename2tikvid(cfgreq.ConfigurationFilename)] = true
		} else {
			relatedComp := request.ClusterType.ConfigType2Component(cfgreq.GetConfigurationType())
			if relatedComp != "" {
				updatedComponent[relatedComp] = true
			}
		}
	}
	for _, compreq := range request.CCRequest.ComponentRequests {
		updatedComponent[compreq.ComponentName] = true
		if compreq.ComponentName == ccproto.SHARED_POOL_COMPONENT_TIKV {
			// mark all tikv group in this resource pool as need update
			meta, err := a.mgmthttpcli.ConfigCenterNgProxy().GetClusterMeta2(ctx, request.LogicalRegion, request.CloudProvider, &ccproto.GetClusterMetaRequest{
				ClusterType: request.ClusterType,
				ClusterId:   request.SharedPoolID,
			})
			if err != nil {
				return fmt.Errorf("get cluster meta failed: %w", err)
			}
			for _, tikvConfigFile := range meta.GetConfigurationFiles() {
				if tikvConfigFile.GetConfigurationType() == ccproto.SHARED_POOL_CONFIG_TIKV {
					tikvID := filename2tikvid(tikvConfigFile.GetConfigurationFilename())
					updatedTikvID[tikvID] = true
				}
			}
		}
	}

	for comp := range updatedComponent {
		switch comp {
		case "tikv":
			// tikv TargetComponent must be added per tikv group, handle it later
			continue
		case ccproto.SHARED_POOL_COMPONENT_TIPROXY:
			tiproxyID, ok, err := sharedPoolTiProxyTargetID(getSharedPool)
			if err != nil {
				return err
			}
			if !ok {
				continue
			}
			request.ClusterSvcReq.Components = append(request.ClusterSvcReq.Components, &clusterng.TargetComponent{
				ComponentType: clusterng.ComponentType_TIPROXY,
				Id:            tiproxyID,
			})
		default:
			request.ClusterSvcReq.Components = append(request.ClusterSvcReq.Components, &clusterng.TargetComponent{
				ComponentType: clusterng.ComponentType(clusterng.ComponentType_value[strings.ToUpper(comp)]),
			})
		}
	}
	for tikvID := range updatedTikvID {
		applyComp := &clusterng.TargetComponent{
			ComponentType: clusterng.ComponentType_TIKV,
			Id:            tikvID,
		}
		request.ClusterSvcReq.Components = append(request.ClusterSvcReq.Components, applyComp)
	}
	return nil
}

func sharedPoolTiProxyTargetID(getSharedPool func() (*mgmthttpcli.SharedPool, error)) (string, bool, error) {
	sharedPool, err := getSharedPool()
	if err != nil {
		return "", false, err
	}

	defaultComponentID := ""
	for _, comp := range sharedPool.ComponentGroups {
		if comp.ComponentType == clusterng.ComponentType_PD.String() {
			defaultComponentID = comp.Id
			break
		}
	}
	if defaultComponentID == "" {
		return "", false, nil
	}

	for _, comp := range sharedPool.ComponentGroups {
		if comp.ComponentType == clusterng.ComponentType_TIPROXY.String() && comp.Id == defaultComponentID {
			return comp.Id, true, nil
		}
	}
	return "", false, nil
}

func filename2tikvid(filename string) string {
	if filename == "tikv" {
		return ""
	}
	return strings.TrimPrefix(filename, "tikv-")
}

func (a *ResourcePoolApplier) Apply(ctx context.Context, actx *ApplyContext) error {
	// chg := actx.Input.Change
	request := actx.ResourcePoolApplierRequest
	logger := log.FromContext(ctx).WithName(a.Name())
	applySemConfig := false

	for _, fileReq := range request.CCRequest.GetConfigurationFileRequests() {
		if fileReq.GetConfigurationType() == ccproto.SHARED_POOL_CONFIG_TIDB_SEM {
			applySemConfig = true
			break
		}
	}

	// 1. apply the file change via the config center through mgmt proxy
	if actx.ApplyMode.Appliable(api.ConfigCenterApply) {
		logger.Info("applying cluster config change via config center mgmt proxy", "req", protojson.Format(request.CCRequest))
		_, err := a.mgmthttpcli.ConfigCenterNgProxy().UpdateClusterMeta2(ctx, request.LogicalRegion, request.CloudProvider, request.CCRequest)
		if err != nil {
			return fmt.Errorf("apply cluster config via config center failed: %w", err)
		}
	} else {
		logger.Info("skip apply via config center cause apply method missed in apply mode")
	}

	// 2. apply the change via the cluster-service-ng through mgmt proxy
	if actx.ApplyMode.Appliable(api.MgmtServiceApply) {
		applyComponents := make([]*clusterng.TargetComponent, 0)
		for _, comp := range request.ClusterSvcReq.Components {
			if skipSharedPoolComponentApplyViaClusterService(request.CCRequest.GetClusterType(), comp.ComponentType) {
				logger.Info("skip apply component via cluster-service-ng", "component", comp.ComponentType.String(), "id", comp.Id)
			} else {
				applyComponents = append(applyComponents, comp)
			}
		}
		request.ClusterSvcReq.Components = applyComponents
		regionServer := a.mgmthttpcli.NextGenRegionServer(request.LogicalRegion, request.CloudProvider)
		if len(request.ClusterSvcReq.Components) == 0 {
			logger.Info("no component rollout needed via cluster-service-ng")
		} else {
			// request.MgmtRequest.StatusId = setStatusResp.StatusId
			reqJSON, _ := json.Marshal(request.ClusterSvcReq) // for debug
			logger.Info("applying cluster config change via mgmt http", "req", string(reqJSON))
			err := regionServer.SetSharedPoolComponentConfigVersion(ctx, request.ClusterSvcReq)
			if err != nil {
				return fmt.Errorf("apply cluster config via cluster-service-ng failed: %w", err)
			}
		}
		if applySemConfig {
			semReq := &clusterng.ApplySemConfigRequest{
				SharedPoolId: request.SharedPoolID,
			}
			logger.Info("applying shared pool sem config via mgmt http", "shared_pool_id", request.SharedPoolID)
			err := regionServer.ApplySemConfig(ctx, semReq)
			if err != nil {
				return fmt.Errorf("apply sem config via cluster-service-ng failed: %w", err)
			}
		}
	} else {
		logger.Info("skip apply via cluster-service-ng cause apply method missed in apply mode")
	}

	// send event
	clusterIDUint, _ := strconv.ParseUint(request.SharedPoolID, 10, 64)
	cluster, err := a.coresvc.Mgmt.GetCluster(ctx, clusterIDUint)
	if err != nil {
		logger.Error(err, "query cluster failed when publishing config change apply event", "cluster_id", request.ClusterId)
	} else {
		displayEv := &evbus.DisplayEvent{
			DisplayEventType: string(eventent.ConfigChangeApplyEvent),
			DisplayStatus:    evbus.EventStatusRunning,
			Severity:         evbus.EventSeverityInfo,
			Summary:          string(eventent.ConfigChangeApplySummary),
			HappenedAtMs:     uint64(time.Now().UnixMilli()),
			Author:           "",
			TenantID:         cluster.OrganizationId,
			ProjectID:        0,
			ClusterID:        clusterIDUint,
			Detail:           protojson.Format(request.CCRequest),
		}
		err = a.eventBus.DBaaSDisplayEvent.Publish(displayEv)
		if err != nil {
			logger.Error(err, "publish config change apply event failed", "cluster_id", request.ClusterId)
		}
	}

	return nil
}

func skipSharedPoolComponentApplyViaClusterService(clusterType ccproto.ClusterType, componentType clusterng.ComponentType) bool {
	if componentType == clusterng.ComponentType_TIDB {
		return true
	}
	// Only shared-pool TiProxyGroup config/image changes require a cluster-service-ng
	// apply so the dataplane CR is reconciled after ConfigCenter metadata changes.
	if componentType == clusterng.ComponentType_TIPROXY {
		return clusterType != ccproto.ClusterType_SHARED_POOL
	}
	return false
}

func (a *ResourcePoolApplier) CheckResult(ctx context.Context, rctx *CheckResultContext) (bool, string, error) {
	chg := rctx.Input.Change

	clusterIDInt, err := strconv.ParseUint(chg.TargetCluster.ClusterID, 10, 64)
	if err != nil {
		return false, "", fmt.Errorf("parse cluster id to int failed: %w", err)
	}

	rp, err := a.coresvc.Mgmt.GetResourcePool(ctx, clusterIDInt)
	if err != nil {
		cluster, err := a.coresvc.Mgmt.GetCluster(ctx, clusterIDInt)
		if err != nil {
			return false, "", fmt.Errorf("get resource pool and cluster failed: %w", err)
		}
		spIDInt, err := strconv.ParseUint(cluster.GetNextGenCluster().GetSharedPoolId(), 10, 64)
		if err != nil {
			return false, "", fmt.Errorf("parse shared pool id to int failed: %w", err)
		}
		rp, err = a.coresvc.Mgmt.GetResourcePool(ctx, spIDInt)
		if err != nil {
			return false, "", fmt.Errorf("get resource pool failed: %w", err)
		}
	}

	if rp.GetResourcePool() == nil {
		return false, "", fmt.Errorf("resource pool is missing")
	}
	logicalRegion, cloudProvider, ok := resourcePoolRegion(rp)
	if !ok {
		return false, "", fmt.Errorf("resource pool region is missing")
	}

	sp, err := a.mgmthttpcli.NextGenRegionServer(logicalRegion, cloudProvider).GetSharedPool(ctx, strconv.FormatUint(rp.GetResourcePool().GetId(), 10))
	if err != nil {
		return false, "", fmt.Errorf("get shared pool failed: %w", err)
	}
	targets, ok := a.resourcePoolCheckResultTargets(ctx, rctx.Input, rp, sp)
	if ok {
		done, reason := targetComponentsAvailable(sp.ComponentGroups, targets.Components)
		if done {
			err := a.publishResourcePoolFinishEvent(ctx, rp.ResourcePool.OwnerId, clusterIDInt)
			if err != nil {
				logger := log.FromContext(ctx).WithName(a.Name())
				logger.Error(err, "publish config change finish event failed", "cluster_id", clusterIDInt)
			}
			return true, "", nil
		}
		return false, reason, nil
	}

	state := sp.Phase
	if state == "AVAILABLE" {
		// send event
		err = a.publishResourcePoolFinishEvent(ctx, rp.ResourcePool.OwnerId, clusterIDInt)
		if err != nil {
			logger := log.FromContext(ctx).WithName(a.Name())
			logger.Error(err, "publish config change finish event failed", "cluster_id", clusterIDInt)
		}
		return true, "", nil
	}

	return false, fmt.Sprintf("cluster state still is %s", state), nil
}

func (a *ResourcePoolApplier) resourcePoolCheckResultTargets(ctx context.Context, input *Input, rp *nextgenv1.ResourcePoolProfile, sp *mgmthttpcli.SharedPool) (*sharedPoolTargetComponents, bool) {
	if input == nil || input.Version != nil || rp.GetResourcePool() == nil {
		return nil, false
	}
	logicalRegion, cloudProvider, ok := resourcePoolRegion(rp)
	if !ok {
		return nil, false
	}

	sharedPoolID := strconv.FormatUint(rp.GetResourcePool().GetId(), 10)
	request := &ResourcePoolApplierRequest{
		ClusterType:   ccproto.ClusterType_SHARED_POOL,
		ClusterId:     sharedPoolID,
		SharedPoolID:  sharedPoolID,
		LogicalRegion: logicalRegion,
		CloudProvider: cloudProvider,
		ClusterSvcReq: &clusterng.SetSharedPoolComponentConfigVersionRequest{
			SharedPoolId: sharedPoolID,
			Components:   []*clusterng.TargetComponent{},
		},
		CCRequest: inputClusterMetaRequest(ccproto.ClusterType_SHARED_POOL, sharedPoolID, input),
	}
	if err := a.setResourcePoolTargetComponents(ctx, request, func() (*mgmthttpcli.SharedPool, error) {
		return sp, nil
	}); err != nil {
		return nil, false
	}
	targets := componentGroupAvailabilityTargets(request.CCRequest.GetClusterType(), request.ClusterSvcReq.Components)

	return &sharedPoolTargetComponents{
		SharedPoolID:  request.SharedPoolID,
		LogicalRegion: request.LogicalRegion,
		CloudProvider: request.CloudProvider,
		Components:    targets,
	}, targetComponentsReliable(targets)
}

func (a *ResourcePoolApplier) publishResourcePoolFinishEvent(ctx context.Context, tenantID, clusterID uint64) error {
	displayEv := &evbus.DisplayEvent{
		DisplayEventType: string(eventent.ConfigChangeFinishEvent),
		DisplayStatus:    evbus.EventStatusSucceeded,
		Severity:         evbus.EventSeverityInfo,
		Summary:          string(eventent.ConfigChangeFinishSummary),
		HappenedAtMs:     uint64(time.Now().UnixMilli()),
		Author:           "",
		TenantID:         tenantID,
		ProjectID:        0,
		ClusterID:        clusterID,
	}
	return a.eventBus.DBaaSDisplayEvent.Publish(displayEv)
}