package cluster

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/dedicated-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/featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (s *serviceImpl) Pause(
	ctx context.Context,
	req *globaldedicatedv1.PauseClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.PauseCluster")
	defer span.End()

	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	var cluster *entity.Cluster
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		var err error
		cluster, err = s.clusterRepo.GetAndLoadComponents(ctx, req.ClusterId)
		if err != nil {
			return err
		}

		switch cluster.State {
		case globaldedicatedv1.Cluster_PAUSED:
			return bizerr.New(bizerr.ErrCodeParams, "cluster is already paused")
		case globaldedicatedv1.Cluster_PAUSING:
			tilog.Info(ctx, "cluster is in PAUSING")
			return nil
		case globaldedicatedv1.Cluster_AVAILABLE:
			// do nothing
		default:
			tilog.Warn(ctx, "cluster can not pause", zap.String("state", cluster.State.String()))
			return bizerr.New(
				bizerr.ErrCodeParams,
				"the cluster can be paused only when it is %s",
				globaldedicatedv1.Cluster_AVAILABLE,
			)
		}

		// Set cluster state to pausing
		cluster.State = globaldedicatedv1.Cluster_PAUSING
		cluster.HealthState = globaldedicatedv1.Cluster_UNHEALTHY

		var pausePlan *entity.ClusterPausePlan
		enableNormalPlan, err := s.featGateCli.CheckFeatureEnabledV2(
			ctx,
			featuregate.NormalPause,
			cluster.OrganizationId,
			cluster.ProjectId,
			cluster.CloudProviderId.String(),
			cluster.RegionId.String(),
		)
		if err != nil {
			return bizerr.New(bizerr.ErrCodeUnknown, "check feature gate failed: %w", err)
		}
		if enableNormalPlan {
			pauseTime := time.Now()
			pausePlan = &entity.ClusterPausePlan{
				Cluster_PausePlan: &globaldedicatedv1.Cluster_PausePlan{
					PausePlan: globaldedicatedv1.Cluster_PausePlan_NORMAL, // only support normal pause yet
					PauseTime: timestamppb.New(pauseTime),
					ScheduledResumeTime: timestamppb.New(
						pauseTime.Add(time.Hour * 24 * time.Duration(s.cfg.BizConfig.NormalPauseNotifyHour)),
					),
				},
			}
			// set cluster pause plan
			cluster.PausePlan = pausePlan.Cluster_PausePlan
		}

		if req.ValidateOnly {
			return nil
		}

		// Save cluster record to db.
		if err := s.clusterRepo.Save(ctx, cluster); err != nil {
			return err
		}

		// save cluster pause plan
		if pausePlan != nil {
			if err := s.pausePlanRepo.Save(ctx, pausePlan, cluster.ClusterId); err != nil {
				return err
			}
		}

		metadataJson, merr := metadata.MarshalRequestMetaFromCtx(ctx)
		if merr != nil {
			return fmt.Errorf("failed to marshal request metadata: %w", merr)
		}

		// Start async reconciliation workflow.
		workFlowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.PauseCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			fmt.Sprintf("%d", cluster.ClusterId),
			map[string]string{
				workflow.KeyTraceId:     workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta: string(metadataJson),
				common.KeyClusterId:     strconv.FormatUint(req.ClusterId, 10),
				common.KeyTargetStatus:  globaldedicatedv1.Cluster_PAUSED.String(),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "pause cluster failed")
	}

	return cluster, nil
}

func (s *serviceImpl) Resume(
	ctx context.Context,
	req *globaldedicatedv1.ResumeClusterRequest,
	scheduledTime *time.Time,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.ResumeCluster")
	defer span.End()

	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	var cluster *entity.Cluster
	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		var err error
		cluster, err = s.clusterRepo.GetAndLoadComponents(ctx, req.ClusterId)
		if err != nil {
			return err
		}

		switch cluster.State {
		case globaldedicatedv1.Cluster_PAUSED:
			// do nothing
		case globaldedicatedv1.Cluster_RESUMING:
			tilog.Info(ctx, "cluster is in RESUMING")
			return nil
		default:
			tilog.Warn(ctx, "cluster can not resume", zap.String("state", cluster.State.String()))
			return bizerr.New(
				bizerr.ErrCodeParams,
				"the cluster can be resumed only when it is %s",
				globaldedicatedv1.Cluster_PAUSED,
			)
		}

		dagIns, err := s.workflow.ListRunningDagInstances(
			ctx,
			[]string{common.ResumeCluster},
			strconv.FormatUint(req.ClusterId, 10),
		)
		if err != nil {
			return bizerr.WrapError(err, "check cluster resume status failed")
		}

		if len(dagIns) > 0 {
			// auto resume
			if scheduledTime != nil {
				tilog.Info(ctx, "cluster has running auto resume workflow")
				return nil
			}

			// manual resume
			// will delete running auto resume workflow
			if err := s.workflow.DeleteDagInstancesWithDagId(ctx, fmt.Sprintf("%d", cluster.ClusterId), common.ResumeCluster); err != nil {
				return err
			}
		}

		// if set scheduled time, not set cluster state to resuming yet
		// set cluster state to resuming
		if scheduledTime == nil {
			cluster.State = globaldedicatedv1.Cluster_RESUMING
		}
		cluster.HealthState = globaldedicatedv1.Cluster_UNHEALTHY

		if req.ValidateOnly {
			return nil
		}

		// if no set scheduled time, use current time
		var pausePlan string
		if scheduledTime == nil {
			scheduledTime = lo.ToPtr(time.Now())
		} else {
			// send scheduled time event
			pausePlan = globaldedicatedv1.Cluster_PausePlan_NORMAL.String() // if scheduled time is nil, it is manual resume
			ev := entity.NewClusterEventBuilder(ctx, cluster, clusterevent.EventTypeScheduledResume).Build()
			if err := s.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, ev); err != nil {
				return err
			}
		}

		// Save cluster record to db.
		if err := s.clusterRepo.Save(ctx, cluster); err != nil {
			return err
		}

		metadataJson, merr := metadata.MarshalRequestMetaFromCtx(ctx)
		if merr != nil {
			return fmt.Errorf("failed to marshal request metadata: %w", merr)
		}

		// Start async reconciliation workflow.
		workFlowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.ResumeCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			fmt.Sprintf("%d", cluster.ClusterId),
			map[string]string{
				workflow.KeyTraceId:        workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:    string(metadataJson),
				workflow.KeyScheduledTime:  scheduledTime.Format(time.RFC3339),
				common.KeyClusterId:        strconv.FormatUint(req.ClusterId, 10),
				common.KeyClusterPausePlan: pausePlan,
				common.KeyTargetStatus:     globaldedicatedv1.Cluster_AVAILABLE.String(),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "resume cluster failed")
	}

	return cluster, nil
}