package cluster

import (
	"context"
	"strconv"

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

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/serverless"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/randstr"
)

// TakeoverCluster picks a prepared shadow cluster from the given shadow pool, resets its
// root password (using the provided passwd, or generating one if empty), and transitions
// the cluster to TakenOver state. The entire operation runs inside a single transaction.
// orgID must match the pool's OrganizationID, otherwise a 403 error is returned.
func (svc *service) TakeoverCluster(
	ctx context.Context,
	shadowPoolID uint64,
	orgID uint64,
	passwd string,
) (*entity.Cluster, error) {
	spanName := "ClusterService.TakeoverCluster"
	ctx, span := serverless.Tracer.Start(ctx, spanName)
	defer span.End()

	// Verify that the caller's org matches the pool's org before acquiring any locks.
	pool, err := svc.shadowPoolRepo.Get(ctx, shadowPoolID)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get shadow pool %d", shadowPoolID)
	}
	if pool.OrganizationID != orgID {
		return nil, bizerr.New(bizerr.ErrCodePermission,
			"access denied to shadow pool %d", shadowPoolID)
	}

	var takenOverCluster *entity.Cluster

	if err := svc.transaction.Transaction(ctx, func(txCtx context.Context) error {
		// Query prepared clusters in the shadow pool (with row-level lock)
		clusters, err := svc.clusterRepo.QueryPreparedShadowClustersForUpdate(txCtx, shadowPoolID, 1)
		if err != nil {
			return bizerr.WrapGormError(err, "failed to query clusters from shadow pool")
		}

		// Find a prepared cluster
		var preparedCluster *entity.Cluster
		for _, cluster := range clusters {
			if cluster.ShadowState == entity.ShadowStatePrepared {
				preparedCluster = cluster
				break
			}
		}

		if preparedCluster == nil {
			return bizerr.New(
				bizerr.ErrCodeRecordNotFound,
				"no prepared cluster available in shadow pool %d",
				shadowPoolID,
			)
		}

		tilog.Info(txCtx, "taking over cluster from shadow pool",
			zap.Uint64("pool_id", shadowPoolID),
			zap.Uint64("cluster_id", preparedCluster.ClusterId))

		// Advance state to TakingOver
		if updateErr := svc.clusterRepo.UpdateShadowState(txCtx,
			preparedCluster.ClusterId,
			entity.ShadowStateTakingOver,
			entity.ShadowStatePrepared); updateErr != nil {
			return bizerr.WrapGormError(updateErr, "failed to update cluster shadow state to TakingOver")
		}

		// Use provided password or generate a random one (same approach as RentCluster in instance.go)
		rootPwd := passwd
		if rootPwd == "" {
			rootPwd = randstr.GenWithCharset(16, randstr.AlphanumericCharset)
		}

		// Reset the root password
		_, err = svc.serverlessCli.ChangeSqlUserPasswd(txCtx, &globalserverlessv1.ChangeSqlUserPasswdRequest{
			ClusterId: strconv.FormatUint(preparedCluster.ClusterId, 10),
			UserName:  "root",
			Passwd:    rootPwd,
		})
		if err != nil {
			return bizerr.WrapError(err, "failed to reset password for taken-over cluster")
		}

		// Advance state to TakenOver
		if updateErr := svc.clusterRepo.UpdateShadowState(txCtx,
			preparedCluster.ClusterId,
			entity.ShadowStateTakenOver,
			entity.ShadowStateTakingOver); updateErr != nil {
			return bizerr.WrapGormError(updateErr, "failed to update cluster shadow state to TakenOver")
		}

		preparedCluster.ShadowState = entity.ShadowStateTakenOver
		takenOverCluster = preparedCluster
		return nil
	}); err != nil {
		return nil, err
	}

	cluster, err := svc.GetCluster(ctx, takenOverCluster.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_FULL)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to get taken-over cluster details")
	}

	tilog.Info(ctx, "successfully took over cluster from shadow pool",
		zap.Uint64("pool_id", shadowPoolID),
		zap.Uint64("cluster_id", takenOverCluster.ClusterId))

	return cluster, nil
}