package server

import (
	"context"

	"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/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

const listSharedPoolPDDeploymentStatusConcurrency = 8

func (s *serverImpl) ListSharedPools(ctx context.Context, sharedPoolIds, dataplaneNames []string) ([]*entity.SharedPool, error) {

	sharedPools, err := s.sharedPoolRepo.List(ctx, &repo.SharedPoolFilter{
		Ids:            sharedPoolIds,
		DataplaneNames: dataplaneNames,
		Deleted:        false,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list clusters: %s", err)
	}

	sharedpoolMap := lo.SliceToMap(sharedPools, func(sharedPool *entity.SharedPool) (string, *entity.SharedPool) {
		return sharedPool.Id, sharedPool
	})

	// get running dag instance
	// add the creating sharedPool which is not saved in db yet
	dagIns, err := s.getRunningDagIns(ctx, []string{dag.PlanCreateSharedPool}, nil, nil, nil, nil)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get running dag instance: %s", err)
	}

	for _, ins := range dagIns {
		sharedPoolId, err := ins.ExecTask.Args.GetString(workflow.SharedPoolIdKey)
		if err != nil {
			tilog.Error(ctx, "failed to get tidb sharedPool id", zap.String("task name", ins.ExecTask.ExecId), zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to get tidb sharedPool id: %s", err)
		}

		if _, ok := sharedpoolMap[sharedPoolId]; ok {
			// sharedPool already exists in db, skip
			continue
		}

		sharedPoolStr, err := ins.ExecTask.Args.GetString(workflow.SharedPoolKey)
		if err != nil {
			tilog.Error(ctx, "failed to get sharedPool", zap.String("task name", ins.ExecTask.ExecId), zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to get sharedPool: %s", err)
		}
		var sharedPool entity.SharedPool
		err = protojson.Unmarshal([]byte(sharedPoolStr), &sharedPool)
		if err != nil {
			tilog.Error(ctx, "failed to unmarshal sharedPool", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to unmarshal sharedPool: %s", err)
		}
		sharedPool.DeployType, _ = ins.ExecTask.Args.GetString(workflow.SharedPoolDeployTypeKey)
		sharedPool.Phase = apiv1.SharedPool_CREATING
		sharedpoolMap[sharedPool.GetId()] = &sharedPool
	}

	pools := lo.Values(sharedpoolMap)
	runningPDDeploymentModeSharedPoolIDs, err := s.listRunningPDDeploymentModeSharedPoolIDs(ctx, lo.Keys(sharedpoolMap))
	if err != nil {
		tilog.Warn(ctx, "failed to get running pd deployment mode dag for shared pool list", zap.Error(err))
	}

	// merge sharedPool status from dag task
	for _, sharedPool := range pools {
		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)
		}
	}

	group, groupCtx := errgroup.WithContext(ctx)
	group.SetLimit(listSharedPoolPDDeploymentStatusConcurrency)
	for _, sharedPool := range pools {
		sharedPool := sharedPool
		_, hasRunningPDDeploymentModeDAG := runningPDDeploymentModeSharedPoolIDs[sharedPool.GetId()]
		group.Go(func() error {
			s.mergePDDeploymentStatusForReadWithRunningDAGBestEffort(groupCtx, sharedPool, hasRunningPDDeploymentModeDAG)
			return nil
		})
	}
	if err := group.Wait(); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to merge pd deployment status: %s", err)
	}
	for _, sharedPool := range pools {
		if err := projectSharedPoolPhase(sharedPool); err != nil {
			return nil, status.Errorf(codes.Internal, "failed to project shared pool phase: %s", err)
		}
	}

	return pools, nil
}

func (s *serverImpl) listRunningPDDeploymentModeSharedPoolIDs(
	ctx context.Context,
	sharedPoolIDs []string,
) (map[string]struct{}, error) {
	runningSharedPoolIDs := make(map[string]struct{})
	if len(sharedPoolIDs) == 0 {
		return runningSharedPoolIDs, nil
	}

	runningDags, err := s.getRunningDagIns(ctx, pdDeploymentModeDAGPlans(), sharedPoolIDs, nil, nil, nil)
	if err != nil {
		return nil, err
	}
	for _, runningDag := range runningDags {
		if runningDag == nil || runningDag.ExecTask == nil || runningDag.ExecTask.Args == nil {
			tilog.Warn(ctx, "skipping pd deployment mode dag with missing task args")
			continue
		}
		sharedPoolID, err := runningDag.ExecTask.Args.GetString(workflow.SharedPoolIdKey)
		if err != nil {
			tilog.Warn(ctx, "skipping pd deployment mode dag without shared pool id", zap.Error(err))
			continue
		}
		runningSharedPoolIDs[sharedPoolID] = struct{}{}
	}
	return runningSharedPoolIDs, nil
}