package server

import (
	"context"
	"fmt"

	"github.com/samber/lo"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

func (s *serverImpl) GetSharedPool(ctx context.Context, sharedPoolId string) (*entity.SharedPool, error) {
	if sharedPoolId == "" {
		return nil, status.Errorf(codes.InvalidArgument, "sharedPoolId is empty")
	}

	sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		if status.Code(err) != codes.NotFound {
			return nil, status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
		}

		// not found, check if it's in creating
		return nil, status.Errorf(codes.NotFound, "sharedPool %s not found", sharedPoolId)
	}

	// merge sharedPool status from dag task
	err = s.mergeSharedPoolStatusFromDagTask(ctx, sharedPool)
	if err != nil {
		tilog.Error(ctx, "failed to merge sharedPool status from dag task", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed to merge sharedPool status from dag task: %s", err)
	}
	if err := s.mergeCassandraProxyStatusFromRepo(ctx, sharedPool); err != nil {
		tilog.Error(ctx, "failed to merge cassandra-proxy status from repo", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed to merge cassandra-proxy status from repo: %s", err)
	}
	s.mergePDDeploymentStatusForReadBestEffort(ctx, sharedPool)
	if err := projectSharedPoolPhase(sharedPool); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to project shared pool phase: %s", err)
	}

	return sharedPool, nil
}

// mergeSharedPoolStatusFromDagTask merge sharedPool status from dag task
// because the dag task is running in the background, we need to merge the status from the dag task
func (s *serverImpl) mergeSharedPoolStatusFromDagTask(ctx context.Context, sharedPool *entity.SharedPool) error {
	err := s.mergeSharedPoolStatusFromDagTaskExceptRestore(ctx, sharedPool)
	if err != nil {
		return fmt.Errorf("failed to merge sharedPool status from dag task except restore: %w", err)
	}

	// restore alreay persist dag status in db
	err = s.mergeSharedPoolStatusFromRestore(ctx, sharedPool)
	if err != nil {
		return fmt.Errorf("failed to merge sharedPool status from dag task restore: %w", err)
	}
	return nil
}

func (s *serverImpl) mergeCassandraProxyStatusFromRepo(ctx context.Context, sharedPool *entity.SharedPool) error {
	if s.cassandraProxyStateRepo == nil {
		return nil
	}
	states, err := s.cassandraProxyStateRepo.ListBySharedPoolID(ctx, sharedPool.GetId())
	if err != nil {
		return err
	}
	stateByClusterID := lo.SliceToMap(states, func(state *entity.CassandraProxyState) (string, *entity.CassandraProxyState) {
		return state.ClusterID, state
	})
	seenClusterIDs := map[string]struct{}{}
	for _, component := range sharedPool.GetComponentGroups() {
		if component.GetComponentType() != apiv1.ComponentType_CASSANDRA_PROXY {
			continue
		}
		state, ok := stateByClusterID[component.GetId()]
		if !ok {
			continue
		}
		mergeCassandraProxyStateToComponent(component, state)
		seenClusterIDs[state.ClusterID] = struct{}{}
	}
	for _, state := range states {
		if _, ok := seenClusterIDs[state.ClusterID]; ok {
			continue
		}
		sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, newCassandraProxyComponent(state, entity.CassandraProxyPhaseToComponentPhase(state.Phase)))
	}
	return nil
}

func mergeCassandraProxyStateToComponent(component *entity.Component, state *entity.CassandraProxyState) {
	component.Phase = entity.CassandraProxyPhaseToComponentPhase(state.Phase)
	component.NodeQuota = state.NodeQuotaEntity
	if component.Labels == nil {
		component.Labels = map[string]string{}
	}
	component.Labels[cassandraProxyHostIdLabelKey] = state.HostID
	component.Labels[cassandraProxyNodeSpecLabelKey] = state.NodeSpec
}

func (s *serverImpl) mergeSharedPoolStatusFromDagTaskExceptRestore(ctx context.Context, sharedPool *entity.SharedPool) error {
	statusAndDagPlanMap := map[string]apiv1.SharedPool_Phase{
		dag.PlanCreateSharedPool:                            apiv1.SharedPool_CREATING,
		dag.PlanDeleteSharedPool:                            apiv1.SharedPool_DELETING,
		dag.PlanUpdateComponent:                             apiv1.SharedPool_MODIFYING,
		dag.PlanApplyComponent:                              apiv1.SharedPool_MODIFYING,
		dag.PlanCreateComponent:                             apiv1.SharedPool_MODIFYING,
		dag.PlanDeleteTiKVGroup:                             apiv1.SharedPool_MODIFYING,
		dag.PlanCreateTiDBCluster:                           apiv1.SharedPool_MODIFYING,
		dag.PlanUpdateTiDBCluster:                           apiv1.SharedPool_MODIFYING,
		dag.PlanDeleteTiDBCluster:                           apiv1.SharedPool_MODIFYING,
		dag.PlanCreateTiDBGroup:                             apiv1.SharedPool_MODIFYING,
		dag.PlanUpdateTiDBGroup:                             apiv1.SharedPool_MODIFYING,
		dag.PlanDeleteTiDBGroup:                             apiv1.SharedPool_MODIFYING,
		dag.PlanSetSharedPoolPDDeploymentModeToMonolithic:   apiv1.SharedPool_MODIFYING,
		dag.PlanSetSharedPoolPDDeploymentModeToMicroservice: apiv1.SharedPool_MODIFYING,
		dag.PlanEnableCoprocessorWorker:                     apiv1.SharedPool_MODIFYING,
		dag.PlanDisableCoprocessorWorker:                    apiv1.SharedPool_MODIFYING,
	}

	for dagPlan, clusterStatus := range statusAndDagPlanMap {

		// break the loop if cluster phase is pausing and status key is paused
		if sharedPool.GetPhase() == apiv1.SharedPool_PAUSING && clusterStatus == apiv1.SharedPool_PAUSED {
			break
		}

		dagIns, err := s.getRunningDagIns(ctx, []string{dagPlan}, []string{sharedPool.GetId()}, nil, nil, nil)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to get running dag instance: %s", err)
		}

		if len(dagIns) == 0 {
			continue
		}

		for runDagName, exec := range dagIns {

			tilog.Info(ctx, "get running dag instance", zap.Any("sharedPoolId", sharedPool.GetId()), zap.Any("dagPlan", runDagName))
			sharedPoolStr, err := exec.ExecTask.Args.GetString(workflow.SharedPoolKey)
			if err != nil || sharedPoolStr == "" {
				sharedPool.Phase = clusterStatus
				tilog.Warn(ctx, "failed to get sharedPool temp instance from dag task, setting phase to default",
					zap.String("sharedPoolId", sharedPool.GetId()),
					zap.String("dagPlan", dagPlan),
					zap.String("clusterStatus", clusterStatus.String()))
				continue
			}

			// create shared pool dag not finished,
			// but the dataplane maybe already created the resources,
			// we not want to set the shared pool phase to creating forever
			if dagPlan == dag.PlanCreateSharedPool && (sharedPool.GetPhase() == apiv1.SharedPool_AVAILABLE || sharedPool.GetPhase() == apiv1.SharedPool_MODIFYING) {
				tilog.Info(ctx, "shared pool create dag is still running, but shared pool is already available, keep the phase unchanged",
					zap.String("sharedPoolId", sharedPool.GetId()))
				continue
			}

			// get sharedPool from dag task
			var tempSharedPool entity.SharedPool
			err = protojson.Unmarshal([]byte(sharedPoolStr), &tempSharedPool)
			if err != nil {
				return status.Errorf(codes.Internal, "failed to unmarshal sharedPool: %s", err)
			}

			if sharedPool.GetPhase() != tempSharedPool.GetPhase() {
				sharedPool.Phase = tempSharedPool.GetPhase()
			}

			sharedPool.Version = tempSharedPool.GetVersion()

			for _, tempComponent := range tempSharedPool.GetComponentGroups() {
				// check if the component is affected by the running DAG
				// Use CheckIsAffectedComponent to prioritize AffectedComponentTypesKey for overlay filtering
				if !workflow.CheckIsAffectedComponent(ctx, tempComponent, exec.ExecTask.Args) {
					continue
				}

				c, ok := lo.Find(sharedPool.GetComponentGroups(), func(item *entity.Component) bool {
					return item.GetId() == tempComponent.GetId() && tempComponent.GetComponentType() == item.GetComponentType()
				})

				if ok {
					// merge component status
					if c.GetPhase() != tempComponent.GetPhase() {
						c.Phase = tempComponent.GetPhase()
						c.NodeQuota = tempComponent.GetNodeQuota()
						c.Storages = tempComponent.GetStorages()
					}
				} else {
					// add new component
					sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, tempComponent)
				}
			}
		}
	}
	return nil
}

func projectSharedPoolPhase(sharedPool *entity.SharedPool) error {
	if sharedPool.GetPhase() == apiv1.SharedPool_DELETING {
		return nil
	}
	if err := entity.ValidatePauseResumeCombination(sharedPool); err != nil {
		return err
	}
	switch sharedPool.PauseResumeState {
	case entity.PauseResumeStateNone:
		return nil
	case entity.PauseResumeStatePausing:
		sharedPool.Phase = apiv1.SharedPool_PAUSING
	case entity.PauseResumeStatePaused:
		sharedPool.Phase = apiv1.SharedPool_PAUSED
	case entity.PauseResumeStateResuming:
		sharedPool.Phase = apiv1.SharedPool_RESUMING
	}
	return nil
}

func (s *serverImpl) mergeSharedPoolStatusFromRestore(ctx context.Context, sharedPool *entity.SharedPool) error {
	anyFullfillingDecision, err := s.scaleManager.AnyFullfillingDecision(ctx, sharedPool.GetId())
	if err != nil {
		return fmt.Errorf("failed to check if there are fullfilling scale decisions: %w", err)
	}
	if !anyFullfillingDecision {
		return nil
	}

	sharedPool.Phase = apiv1.SharedPool_MODIFYING
	for _, component := range sharedPool.GetComponentGroups() {
		if component.GetComponentType() == apiv1.ComponentType_TIKV {
			component.Phase = apiv1.Component_MODIFYING
		}
	}
	tilog.Info(ctx, "set sharedPool phase to modifying since tikv scaling is fulling for restores", zap.Any("sharedPool", sharedPool))
	return nil
}