package cluster

import (
	"context"
	"database/sql"
	"strconv"
	"time"

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

	"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/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

type clusterRepoImpl struct {
	provider db.Provider
}

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

func (i *clusterRepoImpl) Save(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	if cluster == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster entity is nil")
	}

	wdb := i.provider.WriteDB(ctx)
	regionRdb := i.provider.ReadDB(ctx)

	// Query IDs of associated models.
	var regionModels []*schema.Region
	if err := regionRdb.Model(&schema.Region{}).
		Preload("RegionTenants").
		Where("region_name = ?", cluster.RegionId.String()).
		Find(&regionModels).
		Error; err != nil {
		return bizerr.WrapGormError(err, "failed to query region")
	}
	if len(regionModels) == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "region %s is not supported", cluster.RegionId.String())
	}
	regionModel, err := schema.SelectRegionByOrgId(ctx, regionModels, cluster.OrganizationId)
	if err != nil {
		return err
	}

	var providerModel schema.Provider
	if err := wdb.Take(&providerModel, "name = ?", cluster.CloudProviderId.String()).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to query provider")
	}
	var cmekModel *schema.CustomerManagedEncryptionKey
	var cmekID *uint64
	if cluster.CustomerManagedEncryptionKeyId != "" {
		cmekId, err := strconv.ParseUint(cluster.CustomerManagedEncryptionKeyId, 10, 64)
		if err != nil {
			return bizerr.New(bizerr.ErrCodeParams, "cmek id is invalid")
		}
		if err := wdb.Take(&cmekModel, "id = ?", cmekId).Error; err != nil {
			return bizerr.WrapGormError(err, "failed to query cmek")
		}
		cmekID = lo.ToPtr(uint64(cmekModel.ID))
	}

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

	// For clusters with explicit IDs, we need to check if they exist first
	// because GORM's Save() with a non-zero ID tries UPDATE first
	if schemaCluster.ID != 0 {
		var count int64
		if err := wdb.Model(&schema.Cluster{}).Where("id = ?", schemaCluster.ID).Count(&count).Error; err != nil {
			return bizerr.WrapGormError(err, "failed to check cluster existence")
		}
		if count == 0 {
			// Record doesn't exist, use Create instead of Save
			if err := wdb.Omit("EncryptedRootPassword").Create(&schemaCluster).Error; err != nil {
				return bizerr.WrapGormError(err, "failed to create cluster")
			}
			return nil
		}
	}

	// Save cluster won't save the encrypted root password.
	if err := wdb.Omit("EncryptedRootPassword").Save(&schemaCluster).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to save cluster")
	}

	return nil
}

func (i *clusterRepoImpl) Delete(ctx context.Context, cluster *entity.Cluster) bizerr.BizError {
	if cluster.ClusterId == 0 {
		return bizerr.New(bizerr.ErrCodeUnknown, "cluster id is 0")
	}

	if err := metadata.Validate(ctx, &metadata.ExpectRequest[uint64]{
		TenantId:  cluster.OrganizationId,
		ProjectId: cluster.ProjectId,
		ClusterId: cluster.ClusterId,
	}); err != nil {
		return bizerr.WrapError(err, "validate failed")
	}

	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) Query(ctx context.Context, filter *clusterrepo.Filter) ([]*entity.Cluster, bizerr.BizError) {
	ens, err := i.baseQuery(ctx, filter, func(rdb *gorm.DB) *gorm.DB {
		return rdb.
			InnerJoins("Region").
			InnerJoins("Provider").
			Joins("CustomerManagedEncryptionKey")
	})
	if err != nil {
		return nil, err
	}

	return ens, nil
}

func (i *clusterRepoImpl) QueryAndLoadComponents(
	ctx context.Context,
	filter *clusterrepo.Filter,
) ([]*entity.Cluster, bizerr.BizError) {
	ens, err := i.baseQuery(ctx, filter, func(rdb *gorm.DB) *gorm.DB {
		return rdb.
			InnerJoins("Region").
			InnerJoins("Provider").
			Joins("CustomerManagedEncryptionKey").
			Preload("ClusterComponents")
	})
	if err != nil {
		return nil, err
	}

	return ens, nil
}

func (i *clusterRepoImpl) Get(
	ctx context.Context,
	clusterId uint64,
	showDeleted bool,
) (*entity.Cluster, bizerr.BizError) {
	en, err := i.baseGet(ctx, clusterId, showDeleted, func(rdb *gorm.DB) *gorm.DB {
		return rdb.
			InnerJoins("Region").
			InnerJoins("Provider").
			Joins("CustomerManagedEncryptionKey")
	})
	if err != nil {
		return nil, err
	}

	return en, nil
}

func (i *clusterRepoImpl) GetAndLoadComponents(
	ctx context.Context,
	clusterId uint64,
) (*entity.Cluster, bizerr.BizError) {
	en, err := i.baseGet(ctx, clusterId, false, func(rdb *gorm.DB) *gorm.DB {
		return rdb.
			InnerJoins("Region").
			InnerJoins("Provider").
			Joins("CustomerManagedEncryptionKey").
			Preload("ClusterComponents")
	})
	if err != nil {
		return nil, err
	}

	return en, nil
}

// The presence of the baseQuery function alleviates the SQL read amplification issue.
//   - Not preload the cluster components: one join query.
//   - Preload the cluster components: one join query + N queries for the components.
//   - Deeply preload the cluster components:
//     one join query +
//     N components queries +
//     N * M tier unit conditions queries
func (i *clusterRepoImpl) baseQuery(
	ctx context.Context,
	filter *clusterrepo.Filter,
	joinFunc func(*gorm.DB) *gorm.DB,
) ([]*entity.Cluster, bizerr.BizError) {
	if filter == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster filter is nil")
	}

	rdb := i.provider.ReadDB(ctx)
	var schemaClusters []*schema.Cluster

	// Build query.
	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.ProjectIds) > 0 {
		rdb = rdb.Where("`clusters`.`project_id` IN ?", filter.ProjectIds)
	}
	if len(filter.ClusterIds) > 0 {
		rdb = rdb.Where("`clusters`.`id` IN ?", filter.ClusterIds)
	}
	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.CmekIds) > 0 {
		rdb = rdb.Where("`clusters`.`customer_managed_encryption_key_id` IN ?", filter.CmekIds)
	}
	if filter.ShowDeleted {
		rdb = rdb.Unscoped()
	}

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

	for _, schemaCluster := range schemaClusters {
		if err := i.deepPreloadClusterComponents(ctx, schemaCluster, filter.ShowDeleted); err != nil {
			return nil, err
		}
	}

	entities := make([]*entity.Cluster, 0, len(schemaClusters))
	for _, schemaCluster := range schemaClusters {
		e, err := cvtClusterModelToEntity(schemaCluster)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to convert cluster")
		}
		entities = append(entities, e)
	}

	return entities, nil
}

// baseGet function alleviates the SQL read amplification issue. Refer to the [baseQuery] function for details.
func (i *clusterRepoImpl) baseGet(
	ctx context.Context,
	clusterId uint64,
	showDeleted bool,
	joinFunc func(*gorm.DB) *gorm.DB,
) (*entity.Cluster, bizerr.BizError) {
	rdb := i.provider.ReadDB(ctx)
	if showDeleted {
		rdb = rdb.Unscoped()
	}
	var schemaCluster schema.Cluster

	if err := joinFunc(rdb).Take(&schemaCluster, clusterId).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "failed to query cluster, cluster id %d", clusterId)
	}

	if err := i.deepPreloadClusterComponents(ctx, &schemaCluster, showDeleted); err != nil {
		return nil, err
	}

	e, err := cvtClusterModelToEntity(&schemaCluster)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to convert cluster, cluster id %d", clusterId)
	}

	return e, nil
}

func (i *clusterRepoImpl) deepPreloadClusterComponents(
	ctx context.Context,
	cluster *schema.Cluster,
	showDeleted bool,
) bizerr.BizError {
	rdb := i.provider.ReadDB(ctx)

	for i := range cluster.ClusterComponents {
		var (
			rdbTidbGroup *gorm.DB
			rdbStorages  *gorm.DB
		)
		if showDeleted {
			rdbTidbGroup = rdb.Unscoped()
			rdbStorages = rdb.Unscoped()
		} else {
			rdbTidbGroup = rdb
			rdbStorages = rdb
		}

		// load tidb group
		err := rdbTidbGroup.
			Where("`id` = ?", cluster.ClusterComponents[i].WorkPoolId).
			Take(&cluster.ClusterComponents[i].WorkPool).Error
		if err != nil {
			return bizerr.WrapGormError(err, "failed to query work pool, cluster id %d", cluster.ID)
		}

		// load storage
		err = rdbStorages.
			Where("`cluster_component_id` = ?", cluster.ClusterComponents[i].ID).
			Find(&cluster.ClusterComponents[i].Storages).Error
		if err != nil {
			return bizerr.WrapGormError(err, "failed to query storages, cluster id %d", cluster.ID)
		}

		// load component tier unit
		err = rdb.
			Where("`component_tier_units`.`key` = ?", cluster.ClusterComponents[i].ComponentTierUnitKey).
			Joins("ComponentTier").
			InnerJoins("Provider").
			Joins("InstanceType").
			Joins("StorageTemplate").
			Joins("RaftStorageTemplate").
			Preload("TierUnitConditions").
			Take(&cluster.ClusterComponents[i].ComponentTierUnit).Error
		if err != nil {
			return bizerr.WrapGormError(err, "failed to query cluster component, cluster id %d", cluster.ID)
		}
		cluster.ClusterComponents[i].ComponentTierUnit.Provider = cluster.Provider
	}

	return nil
}

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 schemaCluster schema.Cluster
	if err := rdb.Select("encrypted_root_password").Where("id = ?", clusterId).Take(&schemaCluster).Error; err != nil {
		return "", bizerr.WrapGormError(err, "failed to get root password")
	}

	return schemaCluster.EncryptedRootPassword, nil
}

/*
GetBillableClustersCount count the billable clusters(create_finished_at is not null) in this project and between startTime and endTime.
Refer to the chart below, option2, option3, option4, option5 should be considered billable in this time windows.

	option1(❌): created----deleted
	option2(✅): created--------------------deleted
	option3(✅): created----------------------------------------------deleted(or null)
	option4(✅): ---------------------------created----deleted-----------------
	option5(✅): ---------------------------created-------------------deleted(or null)
	option6(❌): -----------------------------------------------------created----deleted(or null)
	             -------------------[start--------------------- end]------------
*/
func (i *clusterRepoImpl) GetBillableClustersCount(
	ctx context.Context,
	projectID uint64,
	startTime, endTime time.Time,
) (uint32, bizerr.BizError) {
	rdb := i.provider.ReadDB(ctx)
	var count int64
	if err := rdb.Model(&schema.Cluster{}).Unscoped().
		Where(`
project_id = @project_id
AND create_finished_at IS NOT NULL
AND created_at <= @end_time
AND (deleted_at IS NULL OR deleted_at >= @start_time)
`, sql.Named("project_id", projectID), sql.Named("start_time", startTime), sql.Named("end_time", endTime)).
		Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "failed to count billable clusters")
	}
	return uint32(count), nil
}