package server

import (
	"context"
	"fmt"
	"strings"

	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/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"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) SetDMComponentConfigOrVersion(
	ctx context.Context,
	sharedPoolId string,
	targetComponents []*entity.TargetComponent,
	scope entity.SetVersionConfigScope,
	force bool,
) error {
	for _, comp := range targetComponents {
		if comp.GetComponentType() != apiv1.ComponentType_DATA_MIGRATION {
			return status.Errorf(codes.InvalidArgument, "invalid component type: %s", comp.GetComponentType())
		}
	}

	switch scope {
	case apiv1.SetSharedPoolComponentConfigVersionRequest_VERSION,
		apiv1.SetSharedPoolComponentConfigVersionRequest_ALL,
		apiv1.SetSharedPoolComponentConfigVersionRequest_CONFIG:
	default:
		return status.Errorf(codes.InvalidArgument, "invalid or unsupported scope: %s", scope.String())
	}
	if err := s.guardSharedPoolMutationByID(ctx, sharedPoolId, sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}

	updateImage := dmShouldUpdateImage(scope)
	tilog.Info(ctx, "set dm component config/version",
		zap.String("sharedPoolId", sharedPoolId),
		zap.String("scope", scope.String()),
		zap.Bool("force", force),
		zap.Int("targetComponentCount", len(targetComponents)),
		zap.Bool("updateImage", updateImage))

	dmClusters, err := s.dmClusterRepo.List(ctx, &repo.DMClusterListFilter{
		SharedPoolIds: []string{sharedPoolId},
		Deleted:       false,
	})
	if err != nil {
		return status.Errorf(codes.Internal, "failed to list dm clusters: %s", err)
	}
	tilog.Info(ctx, "listed dm clusters for config/version update",
		zap.String("sharedPoolId", sharedPoolId),
		zap.Int("dmClusterCount", len(dmClusters)))
	if len(dmClusters) == 0 {
		tilog.Info(ctx, "no dm clusters on shared pool, skip apply",
			zap.String("sharedPoolId", sharedPoolId))
		return nil
	}

	dagIns, err := s.getRunningDagIns(ctx, []string{dag.PlanApplyDMCluster}, []string{sharedPoolId}, nil, nil, nil)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get running dag instances: %s", err)
	}
	if len(dagIns) > 0 && !force {
		return status.Errorf(codes.FailedPrecondition,
			"there is a running operation on dm cluster, please try again later or set force to true")
	}

	var submitErrors []string
	for _, dmCluster := range dmClusters {
		if err := s.applyDMClusterConfigVersion(ctx, sharedPoolId, dmCluster.GetId(), force, updateImage); err != nil {
			submitErrors = append(submitErrors, err.Error())
		}
	}
	if len(submitErrors) > 0 {
		return status.Errorf(codes.Internal, "failed to apply dm cluster config/version: %s",
			strings.Join(submitErrors, "; "))
	}
	return nil
}

func dmShouldUpdateImage(scope entity.SetVersionConfigScope) bool {
	return scope == apiv1.SetSharedPoolComponentConfigVersionRequest_VERSION ||
		scope == apiv1.SetSharedPoolComponentConfigVersionRequest_ALL
}

func (s *serverImpl) applyDMClusterConfigVersion(ctx context.Context, sharedPoolId, dmClusterId string, force, updateImage bool) error {
	var dagExecId string
	var dagSubmitted bool

	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		dmCluster, err := s.dmClusterRepo.Load(ctx, dmClusterId, false)
		if err != nil {
			if status.Code(err) != codes.NotFound {
				return status.Errorf(codes.Internal, "failed to load dm cluster: %s", err)
			}
			return status.Errorf(codes.NotFound, "dm cluster %s not found", dmClusterId)
		}

		if dmCluster.GetSharedPoolId() != sharedPoolId {
			return status.Errorf(codes.PermissionDenied,
				"dm cluster %s does not belong to shared pool %s", dmClusterId, sharedPoolId)
		}

		if !force && dmCluster.GetPhase() != apiv1.DMCluster_AVAILABLE {
			return status.Errorf(codes.FailedPrecondition,
				"dm cluster is not available, current phase: %s", dmCluster.GetPhase().String())
		}

		dmCluster.Phase = apiv1.DMCluster_MODIFYING
		if err := s.dmClusterRepo.Save(ctx, dmCluster); err != nil {
			tilog.Error(ctx, "failed to save dm cluster", zap.String("dmClusterId", dmClusterId), zap.Error(err))
			return status.Errorf(codes.Internal, "failed to save dm cluster: %s", err)
		}

		backend, err := s.dmClusterRepo.GetBackend(ctx, dmClusterId, false)
		if err != nil {
			tilog.Error(ctx, "failed to load dm cluster backend", zap.String("dmClusterId", dmClusterId), zap.Error(err))
			return status.Errorf(codes.Internal, "failed to load dm cluster backend: %s", err)
		}

		dmClusterStr, err := protojson.Marshal(dmCluster)
		if err != nil {
			tilog.Error(ctx, "failed to marshal dm cluster", zap.String("dmClusterId", dmClusterId), zap.Error(err))
			return status.Errorf(codes.Internal, "failed to marshal dm cluster: %s", err)
		}

		dagArgs := actions.ActionArgs{
			workflow.SharedPoolIdKey:  actions.ActionArgString(sharedPoolId),
			workflow.DMClusterIdKey:   actions.ActionArgString(dmClusterId),
			workflow.DMClusterKey:     actions.ActionArgString(string(dmClusterStr)),
			workflow.DMBackendKey:     actions.ActionArgString(backend),
			workflow.DMUpdateImageKey: actions.ActionArgBool(updateImage),
			utils.TraceIdKey:          utils.GetTraceId(ctx),
		}

		dagExecId = utils.GeneratedDagExecId(ctx, fmt.Sprintf("set-dm-config-version-%s-%s", sharedPoolId, dmClusterId))
		if err := s.dagExecutor.Submit(ctx, dag.PlanApplyDMCluster, dagArgs, dagExecId); err != nil {
			tilog.Error(ctx, "failed to submit apply dm cluster dag",
				zap.String("dmClusterId", dmClusterId),
				zap.Error(err))
			return status.Errorf(codes.Internal, "failed to submit apply dm cluster dag: %s", err)
		}
		dagSubmitted = true
		tilog.Info(ctx, "submit apply dm cluster dag for config/version update",
			zap.String("dagExecId", dagExecId),
			zap.String("dmClusterId", dmClusterId),
			zap.String("sharedPoolId", sharedPoolId))
		return nil
	}); err != nil {
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort dag after transaction failure",
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return err
	}
	return nil
}