package cluster

import (
	"context"
	"fmt"
	"strconv"
	"sync"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

const (
	maxBatchMoveSize     = 1000
	batchMoveConcurrency = 10
)

// supportedMoveServicePlans are the NextGen service plans that can be moved between projects.
var supportedMoveServicePlans = []openapicommonv1beta1.ServicePlan{
	openapicommonv1beta1.ServicePlan_Premium,
	openapicommonv1beta1.ServicePlan_BYOC,
	openapicommonv1beta1.ServicePlan_Essential_V2,
}

// MoveCluster moves a single NextGen-backed cluster from its current project
// to targetProjectId within the same org.
// If displayName is non-empty it is used as the cluster's new display name; the
// uniqueness check in the target project is performed against that name.
// If displayName is empty the cluster's existing display name is kept and the
// uniqueness check uses it as-is.
func (svc *service) MoveCluster(
	ctx context.Context,
	orgId, clusterId, targetProjectId uint64,
	displayName string,
) (*entity.Cluster, error) {
	ctx, span := tracer.Start(ctx, "NextGenClusterService.MoveTiDB")
	defer span.End()

	var movedCluster *entity.Cluster

	if err := svc.transaction.Transaction(ctx, func(txCtx context.Context) error {
		cluster, err := svc.clusterRepo.GetCluster(txCtx, clusterId, false)
		if err != nil {
			return bizerr.WrapGormError(err, "failed to get cluster")
		}

		if err := validateMoveCluster(cluster, orgId); err != nil {
			return err
		}

		if cluster.ProjectId == targetProjectId {
			movedCluster = cluster
			return nil
		}

		// Capture the old project ID before any update so it can be carried in the event.
		oldProjectId := cluster.ProjectId

		// Determine the effective display name: use the caller-supplied name when
		// provided, otherwise keep the cluster's existing display name.
		effectiveName := cluster.DisplayName
		if displayName != "" {
			effectiveName = displayName
		}

		// Enforce the same-project uniqueness constraint: a cluster's display name
		// must be unique within the target project.
		count, err := svc.clusterRepo.CountDisplayNameUnderProject(txCtx, targetProjectId, effectiveName)
		if err != nil {
			return bizerr.WrapGormError(err, "failed to check cluster display name in target project")
		}
		if count > 0 {
			return bizerr.New(bizerr.ErrCodeDuplicateDisplayName, bizerr.ErrMsgDupDisplayName)
		}

		if err := svc.clusterRepo.UpdateProjectId(txCtx, clusterId, targetProjectId); err != nil {
			return bizerr.WrapGormError(err, "failed to update cluster project_id")
		}

		// Update the display name when the caller explicitly supplied a new one.
		if displayName != "" {
			if err := svc.clusterRepo.UpdateDisplayName(txCtx, clusterId, displayName); err != nil {
				return bizerr.WrapGormError(err, "failed to update cluster display name")
			}
			cluster.DisplayName = displayName
		}

		if err := svc.accountCli.UpsertClusterRelations(txCtx, clusterId, orgId, targetProjectId); err != nil {
			tilog.Error(
				txCtx,
				"failed to upsert cluster relations in account service after project move",
				zap.Uint64("orgId", orgId),
				zap.Uint64("clusterId", clusterId),
				zap.Uint64("targetProjectId", targetProjectId),
				zap.Error(err),
			)
		}

		// Publish a ClusterProjectChanged event so downstream services are notified.
		if err := svc.publishProjectChangedEvent(txCtx, cluster, oldProjectId, targetProjectId); err != nil {
			return err
		}

		cluster.ProjectId = targetProjectId
		movedCluster = cluster
		return nil
	}); err != nil {
		return nil, err
	}

	// Synchronize the project_id to nextgen-global-service after the DB transaction
	// has committed. Failure is non-fatal but logged for reconciliation.
	if _, syncErr := svc.nextGenCli.UpdateClusterProject(ctx, &globalngv1.UpdateClusterProjectRequest{
		ClusterId: clusterId,
		ProjectId: targetProjectId,
	}); syncErr != nil {
		tilog.Error(ctx,
			"failed to sync project_id to nextgen-global-service after project move",
			zap.Uint64("clusterId", clusterId),
			zap.Uint64("targetProjectId", targetProjectId),
			zap.Error(syncErr),
		)
	}

	return movedCluster, nil
}

// BatchMoveClusters moves up to maxBatchMoveSize NextGen-backed clusters from
// sourceProjectId to targetProjectId within the same org.
// Clusters are moved concurrently (up to batchMoveConcurrency at a time); failures are
// collected and returned but do not prevent other clusters from being processed.
func (svc *service) BatchMoveClusters(
	ctx context.Context,
	orgId, sourceProjectId, targetProjectId uint64,
	servicePlans []string,
) (total int32, successIds, failedIds []string, err error) {
	ctx, span := tracer.Start(ctx, "NextGenClusterService.BatchMoveTiDBs")
	defer span.End()

	filter := &repository.ClusterFilter{
		OrganizationIds: []uint64{orgId},
		ProjectIds:      []uint64{sourceProjectId},
		DeployType:      entity.DeployTypeNextGen,
	}

	if len(servicePlans) > 0 {
		filter.ServicePlans = servicePlans
	} else {
		filter.ServicePlans = lo.Map(supportedMoveServicePlans, func(p openapicommonv1beta1.ServicePlan, _ int) string {
			return p.String()
		})
	}

	clusters, err := svc.clusterRepo.QueryClusterByFilterAndPage(ctx, filter, 0, maxBatchMoveSize)
	if err != nil {
		return 0, nil, nil, bizerr.WrapGormError(err, "failed to query clusters for batch move")
	}

	var (
		mu   sync.Mutex
		wg   sync.WaitGroup
		sem  = make(chan struct{}, batchMoveConcurrency)
		sIds []string
		fIds []string
	)

	for _, c := range clusters {
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			wg.Wait()
			return int32(len(clusters)), sIds, fIds, ctx.Err()
		}
		wg.Add(1)
		go func(c *entity.Cluster) {
			defer wg.Done()
			defer func() { <-sem }()

			idStr := fmt.Sprintf("%d", c.ClusterId)

			if err := validateMoveCluster(c, orgId); err != nil {
				tilog.Warn(ctx, "skipping cluster that failed validation",
					zap.Uint64("cluster_id", c.ClusterId), zap.Error(err))
				mu.Lock()
				fIds = append(fIds, idStr)
				mu.Unlock()
				return
			}

			_, moveErr := svc.MoveCluster(ctx, orgId, c.ClusterId, targetProjectId, "")
			if moveErr != nil {
				tilog.Error(ctx, "failed to move cluster in batch",
					zap.Uint64("cluster_id", c.ClusterId), zap.Error(moveErr))
			}
			mu.Lock()
			if moveErr != nil {
				fIds = append(fIds, idStr)
			} else {
				sIds = append(sIds, idStr)
			}
			mu.Unlock()
		}(c)
	}

	wg.Wait()
	return int32(len(clusters)), sIds, fIds, nil
}

// validateMoveCluster checks that the cluster is eligible for a project move.
// Only NextGen-backed clusters are supported by this service.
func validateMoveCluster(cluster *entity.Cluster, orgId uint64) error {
	if cluster.OrganizationId != orgId {
		return bizerr.New(bizerr.ErrCodePermission, "cluster does not belong to the specified org")
	}

	if cluster.DeployType != entity.DeployTypeNextGen {
		return bizerr.New(bizerr.ErrCodeParams,
			"cluster migration via the nextgen service is only supported for NextGen clusters")
	}

	supported := lo.ContainsBy(supportedMoveServicePlans, func(p openapicommonv1beta1.ServicePlan) bool {
		return p == cluster.ServicePlan
	})
	if !supported {
		return bizerr.New(bizerr.ErrCodeParams,
			"cluster service plan %s is not supported for migration; only Premium, BYOC and Essential_V2 are supported",
			cluster.ServicePlan.String())
	}

	return nil
}

// publishProjectChangedEvent writes a ClusterProjectChanged event to the WAL so that
// downstream services are notified about the cluster being reassigned to a new project.
func (svc *service) publishProjectChangedEvent(
	ctx context.Context,
	cluster *entity.Cluster,
	oldProjectId uint64,
	newProjectID uint64,
) error {
	var (
		operatorID       uint64
		email            string
		openAPIPublicKey string
	)

	if reqMeta, err := metadata.ExtractRequestMeta(ctx); err != nil {
		tilog.Warn(ctx, "publishProjectChangedEvent: extract request meta failed", zap.Error(err))
	} else {
		if id, err := strconv.ParseUint(reqMeta.UserID, 10, 64); err == nil {
			operatorID = id
		}
		email = reqMeta.Email
		openAPIPublicKey = reqMeta.ApiPublicKey
	}

	ev := &clusterevent.ClusterEvent{
		TenantID:         cluster.OrganizationId,
		ProjectID:        newProjectID,
		ClusterID:        cluster.ClusterId,
		TraceID:          clusterevent.TraceIDFromContext(ctx),
		OperatorID:       operatorID,
		UserEmail:        email,
		OpenAPIPublicKey: openAPIPublicKey,
		EventType:        clusterevent.EventTypeProjectChanged,
		ServicePlan:      cluster.ServicePlan.String(),
		ClusterType:      clusterevent.ClusterTypeTidbxInstance,
		ExtraInfo: &clusterevent.EventExtraInfo{
			OldProjectID: oldProjectId,
		},
	}

	if err := svc.eventCommitter.Write(ctx, clusterevent.TopicClusterEvent, ev); err != nil {
		return bizerr.WrapError(err, "write ClusterProjectChanged event")
	}
	return nil
}