package cluster

import (
	"context"
	"errors"
	"fmt"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"gorm.io/gorm"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

// GetCluster merges mgmt DB metadata with global dedicated service data.
//
// view is used only for Local early return; Basic and Full both call global and always
// fetched Regional details before globalMetaOnly was added. globalMetaOnly is a separate
// opt-in (→ ignore_details) because repurposing view to skip Regional would change
// behavior for existing BasicView callers—intentionally avoided to limit regression risk.
func (svc *service) GetCluster(
	ctx context.Context,
	clusterId uint64,
	view entity.ClusterView,
	includeDeleted bool,
	globalMetaOnly bool,
) (*entity.Cluster, error) {
	ctx, span := tracer.Start(ctx, "ClusterService.GetCluster")
	defer span.End()

	cluster := &entity.Cluster{}

	if clusterId == 0 {
		return nil, errors.New("cluster id is invalid")
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		var err error
		cluster, err = svc.clusterRepo.GetCluster(ctx, clusterId, includeDeleted)
		if err != nil {
			return fmt.Errorf("failed to get cluster record: %w", err)
		}

		if cluster.DeployType != entity.DeployTypeDedicated {
			tilog.Warn(ctx, "cluster deploy type is not dedicated", zap.Uint64("cluster_id", clusterId))
			return gorm.ErrRecordNotFound
		}

		if err = svc.hydrateClusterTags(ctx, []*entity.Cluster{cluster}); err != nil {
			return fmt.Errorf("failed to get cluster tags: %w", err)
		}

		if view == entity.ClusterViewClusterViewLocal {
			return nil
		}

		runningDagInstances, err := svc.workflow.ListRunningDagInstances(ctx, []string{
			constant.DagDedicatedCreateCluster,
			constant.DagDedicatedRestoreCluster,
		}, fmt.Sprintf("%d", clusterId))
		if err != nil {
			tilog.Error(ctx, "list running dag instances failed", zap.Error(err))
		}

		ignoreDetails := globalMetaOnly
		dedicatedCluster, err := svc.dedicatedCli.GetCluster(ctx, &globaldedicatedv1.GetClusterRequest{
			ClusterId:     clusterId,
			IgnoreDetails: &ignoreDetails,
		})
		if err != nil {
			var berr bizerr.BizError
			if ok := errors.As(err, &berr); ok && berr != nil {
				if berr.ErrCode() == bizerr.ErrCodeRecordNotFound {
					// If dedicated cluster not found in global service
					// Mabye the cluster is creating in mgmt workflow
					// Return cluster with mgmt workflow cluster info
					tilog.Warn(ctx, "dedicated cluster not found", zap.Uint64("cluster_id", clusterId), zap.Error(berr))
					if len(runningDagInstances) != 0 {
						tilog.Warn(ctx, "dedicated cluster not found, but running dag instance found", zap.Uint64("cluster_id", clusterId))
						cluster = mergeInWorkflowState(ctx, cvtDedicatedCluster(cluster, defaultDedicatedCluster(cluster)), runningDagInstances)
						return nil
					}
					tilog.Warn(ctx, "dedicated cluster not found and no running dag instance", zap.Uint64("cluster_id", clusterId))
				}
			}
			return err
		}
		cluster = cvtDedicatedCluster(cluster, dedicatedCluster)

		return nil
	}); err != nil {
		return cluster, err
	}

	return cluster, nil
}