package cluster

import (
	"context"
	"strconv"

	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/db"
	"go.opentelemetry.io/otel"
	"gorm.io/gorm"

	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

var tracer = otel.Tracer("cluster_repo_impl")

func NewClusterRepo(provider db.Provider) clusterrepo.Repo {
	return &clusterRepoImpl{
		provider: provider,
	}
}

type clusterRepoImpl struct {
	provider db.Provider
}

func (i *clusterRepoImpl) Create(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "clusterRepo.Create")
	defer span.End()
	if cluster == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster entity is nil")
	}

	wdb := i.provider.WriteDB(ctx)
	var providerModel schema.Provider
	if err := wdb.Take(&providerModel, "id = ?", cluster.CloudProviderId).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to query provider")
	}

	schemaCluster, err := ConvertClusterEntityToModel(cluster, uint64(providerModel.ID))
	if err != nil {
		return bizerr.WrapError(err, "failed to convert cluster entity to model")
	}
	if err := wdb.Create(schemaCluster).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to create cluster")
	}
	return nil
}

func (i *clusterRepoImpl) Save(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "clusterRepo.Save")
	defer span.End()
	if cluster == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster entity is nil")
	}
	wdb := i.provider.WriteDB(ctx)
	var providerModel schema.Provider
	if err := wdb.Take(&providerModel, "id = ?", cluster.CloudProviderId).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to query provider")
	}

	schemaCluster, err := ConvertClusterEntityToModel(cluster, uint64(providerModel.ID))
	if err != nil {
		return bizerr.WrapError(err, "failed to convert cluster entity to model")
	}

	// Save cluster won't save the encrypted root password.
	return common.SaveWithVersioning(wdb, schemaCluster, []string{"EncryptedRootPassword"})
}

func (i *clusterRepoImpl) Delete(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "clusterRepo.Delete")
	defer span.End()

	if cluster.GetClusterId() == 0 {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster id is empty")
	}

	wdb := i.provider.WriteDB(ctx)
	if err := wdb.Delete(&schema.Cluster{}, cluster.ClusterId).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to delete cluster")
	}
	return nil
}

func (i *clusterRepoImpl) Get(
	ctx context.Context,
	clusterId uint64,
	showDeleted bool,
) (*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "clusterRepo.Get")
	defer span.End()
	if clusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster id is empty")
	}

	rdb := i.provider.ReadDB(ctx)
	if showDeleted {
		rdb = rdb.Unscoped()
	}

	var modelCluster schema.Cluster
	rdb = rdb.Model(&schema.Cluster{}).InnerJoins("Provider").InnerJoins("Region").
		Where("`clusters`.`id` = ?", clusterId)

	if err := rdb.Take(&modelCluster).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get cluster")
	}

	// Create a new db session.
	rdb = i.provider.ReadDB(ctx)
	if showDeleted {
		rdb = rdb.Unscoped()
	}

	dataplaneId, err := enrichDataplaneId(rdb, modelCluster.SharedPoolId)
	if err != nil {
		return nil, err
	}

	if modelCluster.RegionId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "region id is empty for cluster")
	}

	return convertClusterModelToEntity(&modelCluster, dataplaneId)
}

func (i *clusterRepoImpl) SetEncryptedRootPassword(
	ctx context.Context,
	clusterId uint64,
	rootPassword string,
) bizerr.BizError {
	if clusterId == 0 {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster id is 0")
	}

	wdb := i.provider.WriteDB(ctx)
	if err := wdb.Model(&schema.Cluster{}).Where("id = ?", clusterId).Update("encrypted_root_password", rootPassword).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to set root password")
	}

	return nil
}

func (i *clusterRepoImpl) GetEncryptedRootPassword(
	ctx context.Context,
	clusterId uint64,
) (string, bizerr.BizError) {
	if clusterId == 0 {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "cluster id is 0")
	}

	rdb := i.provider.ReadDB(ctx)
	var modelCluster schema.Cluster
	if err := rdb.Model(&schema.Cluster{}).Where("id = ?", clusterId).Take(&modelCluster).Error; err != nil {
		return "", bizerr.WrapGormError(err, "failed to get encrypted root password")
	}

	return modelCluster.EncryptedRootPassword, nil
}

func (i *clusterRepoImpl) Query(
	ctx context.Context,
	filter *clusterrepo.Filter,
) ([]*entity.Cluster, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "clusterRepo.Query")
	defer span.End()
	if filter == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "filter is nil")
	}

	rdb := i.provider.ReadDB(ctx)
	if filter.ShowDeleted {
		rdb = rdb.Unscoped()
	}

	rdb = rdb.Model(&schema.Cluster{}).InnerJoins("Provider").InnerJoins("Region")
	var modelClusters []schema.Cluster
	if len(filter.CloudProviderNames) > 0 {
		cloudProviders := lo.Map(
			filter.CloudProviderNames,
			func(cloudProvider entity.CloudProviderName, index int) string {
				return cloudProvider.String()
			},
		)
		rdb = rdb.Where("`Provider`.`name` IN ?", cloudProviders)
	}
	if len(filter.RegionNames) > 0 {
		rdb = rdb.Where(
			"`Region`.`region_name` IN ?",
			lo.Map(filter.RegionNames, func(region entity.RegionName, index int) string {
				return region.String()
			}),
		)
	}
	if len(filter.LogicalRegionNames) > 0 {
		rdb = rdb.Where("`Region`.`name` IN ?", filter.LogicalRegionNames)
	}
	if len(filter.OrganizationIds) > 0 {
		rdb = rdb.Where("`clusters`.`tenant_id` IN ?", filter.OrganizationIds)
	}
	if len(filter.ClusterIds) > 0 {
		rdb = rdb.Where("`clusters`.`id` IN ?", filter.ClusterIds)
	}
	if len(filter.SharedPoolIds) > 0 {
		sharedPoolIDs := lo.Map(filter.SharedPoolIds, func(id uint64, _ int) string {
			return strconv.FormatUint(id, 10)
		})
		rdb = rdb.Where("`clusters`.`shared_pool_id` IN ?", sharedPoolIDs)
	}
	if len(filter.States) > 0 {
		states := lo.Map(filter.States, func(state entity.State, index int) string {
			return state.String()
		})
		rdb = rdb.Where("`clusters`.`state` IN ?", states)
	}
	if len(filter.ExcludeStates) > 0 {
		excludeStates := lo.Map(filter.ExcludeStates, func(state entity.State, index int) string {
			return state.String()
		})
		rdb = rdb.Where("`clusters`.`state` NOT IN ?", excludeStates)
	}

	if len(filter.ServicePlans) > 0 {
		servicePlans := lo.Map(filter.ServicePlans, func(plan entity.ServicePlan, index int) string {
			return plan.String()
		})
		rdb = rdb.Where("`clusters`.`service_plan` IN ?", servicePlans)
	}

	if err := rdb.Find(&modelClusters).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "failed to rdb clusters")
	}

	// Create a new db session.
	rdb = i.provider.ReadDB(ctx)
	if filter.ShowDeleted {
		rdb = rdb.Unscoped()
	}

	dpMap, err := enrichDataplaneIds(rdb, modelClusters)
	if err != nil {
		return nil, err
	}

	var clusters []*entity.Cluster
	for idx := range modelClusters {
		modelCluster := &modelClusters[idx]
		if modelCluster.RegionId == 0 {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "region id is empty for cluster")
		}

		cluster, err := convertClusterModelToEntity(modelCluster, dpMap[modelCluster.SharedPoolId])
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to convert cluster model to entity")
		}
		clusters = append(clusters, cluster)
	}
	return clusters, nil
}

func (i *clusterRepoImpl) CountLiveBySharedPoolIDs(
	ctx context.Context,
	sharedPoolIDs []uint64,
) (map[uint64]int64, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "clusterRepo.CountLiveBySharedPoolIDs")
	defer span.End()

	counts := make(map[uint64]int64, len(sharedPoolIDs))
	if len(sharedPoolIDs) == 0 {
		return counts, nil
	}

	ids := lo.Map(sharedPoolIDs, func(id uint64, _ int) string {
		return strconv.FormatUint(id, 10)
	})
	var rows []struct {
		SharedPoolID string `gorm:"column:shared_pool_id"`
		Count        int64  `gorm:"column:count"`
	}
	if err := i.provider.ReadDB(ctx).
		Model(&schema.Cluster{}).
		Select("shared_pool_id, COUNT(*) AS count").
		Where("shared_pool_id IN ?", ids).
		Group("shared_pool_id").
		Scan(&rows).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "count live clusters by shared pool")
	}

	for _, row := range rows {
		id, err := strconv.ParseUint(row.SharedPoolID, 10, 64)
		if err != nil {
			return nil, bizerr.WrapError(err, "parse shared pool id from cluster count")
		}
		counts[id] = row.Count
	}
	return counts, nil
}

func (i *clusterRepoImpl) CountBySharedPoolID(ctx context.Context, sharedPoolId uint64) (int64, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "clusterRepo.CountBySharedPoolID")
	defer span.End()

	if sharedPoolId == 0 {
		return 0, bizerr.New(bizerr.ErrCodeParams, "shared pool id is empty")
	}

	rdb := i.provider.ReadDB(ctx)
	sharedPoolIdStr := strconv.FormatUint(sharedPoolId, 10)

	var count int64
	if err := rdb.Model(&schema.Cluster{}).
		Where("`clusters`.`shared_pool_id` = ?", sharedPoolIdStr).
		Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "failed to count clusters by shared pool id")
	}

	return count, nil
}

func (i *clusterRepoImpl) UpdateProjectId(ctx context.Context, clusterId, projectId uint64) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "clusterRepo.UpdateProjectId")
	defer span.End()

	wdb := i.provider.WriteDB(ctx)
	if err := wdb.Model(&schema.Cluster{}).
		Where("id = ?", clusterId).
		Update("project_id", projectId).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to update cluster project_id")
	}
	return nil
}

// enrichDataplaneId looks up the dataplane ID for a single cluster's shared pool.
func enrichDataplaneId(rdb *gorm.DB, sharedPoolId string) (string, bizerr.BizError) {
	if sharedPoolId == "" {
		return "", nil
	}
	var dataplaneId string
	err := rdb.Model(&schema.ResourcePool{}).
		Select("dataplane_id").
		Where("id = CAST(? AS UNSIGNED)", sharedPoolId).
		Scan(&dataplaneId).Error
	if err != nil {
		return "", bizerr.WrapGormError(err, "failed to get resource pool dataplane_id")
	}
	return dataplaneId, nil
}

// enrichDataplaneIds returns a map from sharedPoolId to dataplaneId for a batch of clusters.
func enrichDataplaneIds(rdb *gorm.DB, clusters []schema.Cluster) (map[string]string, bizerr.BizError) {
	poolIds := make([]string, 0, len(clusters))
	for i := range clusters {
		if clusters[i].SharedPoolId != "" {
			poolIds = append(poolIds, clusters[i].SharedPoolId)
		}
	}
	if len(poolIds) == 0 {
		return map[string]string{}, nil
	}

	type rpRow struct {
		ID          uint64
		DataplaneId string
	}
	var rows []rpRow
	err := rdb.Model(&schema.ResourcePool{}).
		Select("id, dataplane_id").
		Where("id IN (?)",
			lo.Map(poolIds, func(s string, _ int) any { return s }),
		).
		Scan(&rows).Error
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to batch get resource pool dataplane_ids")
	}

	dpMap := make(map[string]string, len(rows))
	for _, r := range rows {
		dpMap[strconv.FormatUint(r.ID, 10)] = r.DataplaneId
	}
	return dpMap, nil
}