package action

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

	"github.com/tidbcloud/fastflow/pkg/entity/run"
	clusterdedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	PollClusterUntilSynced = "poll_regional_cluster_util_synced"

	// SkipWaitSynced is specified by common.KeySkipWaitSynced
	SkipWaitSynced = "true"

	KeyInsufficientEventSendTime = "insufficient_event_send_time"
)

type PollUntilSyncedAction struct {
	regionalCli    regionserver.Client
	clusterRepo    clusterrepo.Repo
	regionRepo     region.Repo
	eventCommitter eventbus.WALCommitter

	// Timeout configuration for handling insufficient resource conditions.
	// Refer to feature spec https://pingcap.feishu.cn/wiki/AWo6wdrNiiJa9Jkbr4rc51xMntB.
	insufficientResourceTimeout time.Duration
}

func NewPollUntilSyncedAction(
	regionalCli regionserver.Client,
	clusterRepo clusterrepo.Repo,
	regionRepo region.Repo,
	committer eventbus.WALCommitter,
	insufficientResourceTimeoutMinutes time.Duration,
) *PollUntilSyncedAction {
	return &PollUntilSyncedAction{
		regionalCli:                 regionalCli,
		clusterRepo:                 clusterRepo,
		regionRepo:                  regionRepo,
		eventCommitter:              committer,
		insufficientResourceTimeout: insufficientResourceTimeoutMinutes,
	}
}

func (p *PollUntilSyncedAction) Name() string {
	return PollClusterUntilSynced
}

func (p *PollUntilSyncedAction) ParameterNew() any {
	return &common.ClusterIdParameter{}
}

func (p *PollUntilSyncedAction) Run(ctx run.ExecuteContext, params any) error {
	cctx, span := workflow.GenerateContext(ctx)

	clusterId, err := common.GetClusterId(params)
	if err != nil {
		tilog.Error(cctx, "failed to get cluster id", zap.Error(err))
		return err
	}

	tilog.Info(cctx, "start to poll regional cluster until synced", zap.Uint64("cluster_id", clusterId))
	defer func() {
		tilog.Info(cctx, "end to poll regional cluster until synced", zap.Uint64("cluster_id", clusterId))
		span.End()
	}()

	if v, ok := ctx.GetVar(common.KeySkipWaitSynced); ok && v == SkipWaitSynced {
		tilog.Info(cctx, "skip wait synced action")
		return nil
	}

	cluster, err := p.clusterRepo.Get(cctx, clusterId, false)
	if err != nil {
		tilog.Error(cctx, "failed to get cluster", zap.Error(err))
		return err
	}

	isReady, err := p.check(cctx, ctx, cluster)
	if err != nil {
		tilog.Error(cctx, "failed to check cluster", zap.Error(err))
		return err
	}

	if !isReady {
		tilog.Warn(cctx, "cluster is not ready", zap.Uint64("cluster_id", clusterId))
		return fmt.Errorf("cluster is not ready, cluster id: %d", clusterId)
	}

	if v, ok := ctx.GetVar(common.KeyNewCluster); ok && v == "true" {
		preparedEv := entity.NewClusterEventBuilder(cctx, cluster, clusterevent.EventTypePrepared).Build()
		launching := entity.NewClusterEventBuilder(cctx, cluster, clusterevent.EventTypeLaunching).Build()
		if err := p.eventCommitter.BatchWrite(cctx, clusterevent.TopicClusterEvent, []clusterevent.Event{preparedEv, launching}); err != nil {
			tilog.Error(cctx, "failed to publish cluster event", zap.Uint64("cluster_id", clusterId), zap.Error(err))
			return err
		}
	}

	return nil
}

func (p *PollUntilSyncedAction) check(
	cctx context.Context,
	ctx run.ExecuteContext,
	cluster *entity.Cluster,
) (isReady bool, err error) {
	regionEntity, err := p.regionRepo.SelectRegionById(cctx, cluster.RegionId, cluster.GetOrganizationId())
	if err != nil {
		tilog.Error(cctx, "failed to find region", zap.Error(err))
		return false, err
	}

	regionCluster, err := p.regionalCli.GetCluster(cctx, regionEntity, &regionaldedicatedv1.GetClusterRequest{
		ClusterId:      strconv.FormatUint(cluster.ClusterId, 10),
		ShowStatusOnly: true,
	})
	if err != nil {
		tilog.Error(cctx, "failed to get cluster", zap.Error(err))
		return false, err
	}

	if regionCluster.ClusterPhase == clusterdedicatedv1.Cluster_SYNCED {
		tilog.Info(cctx, "regional cluster is synced", zap.String("cluster_phase", regionCluster.ClusterPhase.String()))
		return true, nil
	}

	tilog.Warn(
		cctx,
		"regional cluster is not synced yet",
		zap.String("cluster_phase", regionCluster.ClusterPhase.String()),
	)

	// Send insufficient resource event if necessary.
	if regionserver.IsClusterVMInsufficient(regionCluster, p.insufficientResourceTimeout) {
		// Ensure the insufficient resource event is sent only once.
		if _, ok := ctx.ShareData().Get(KeyInsufficientEventSendTime); !ok {
			// Committing event and setting share data in one transaction is hard for current workflow implementation.
			// So we commit the event first, and then set the share data. This ensures at-least once delivery, but could cause duplicated event.
			if err := p.eventCommitter.Write(
				cctx,
				clusterevent.TopicClusterEvent,
				entity.NewClusterEventBuilder(cctx, cluster, clusterevent.EventTypeInsufficientResource).Build(),
			); err != nil {
				tilog.Error(
					cctx,
					"failed to publish cluster event",
					zap.Uint64("cluster_id", cluster.ClusterId),
					zap.Error(err),
				)
				return false, err
			}
			ctx.ShareData().Set(KeyInsufficientEventSendTime, time.Now().Format(time.RFC3339))
		}
	}

	return false, nil
}