package resourcepool

import (
	"bytes"
	"context"
	"errors"
	"maps"
	"time"

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

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	resourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

var tracer = otel.Tracer("resourcepool_repo_impl")

type resourcePoolRepoImpl struct {
	provider db.Provider
}

func NewResourcePoolRepo(provider db.Provider) resourcepoolrepo.Repo {
	return &resourcePoolRepoImpl{
		provider: provider,
	}
}

// Create inserts a new ResourcePool record into the database.
func (i *resourcePoolRepoImpl) Create(ctx context.Context, resourcePool *entity.ResourcePool) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.Create")
	defer span.End()
	if resourcePool == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "resourcePool entity is nil")
	}

	wdb := i.provider.WriteDB(ctx)

	var providerModel schema.Provider
	if err := wdb.Take(&providerModel, "id = ?", resourcePool.CloudProviderId).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to query provider")
	}
	var regionModel schema.Region
	if err := wdb.Take(&regionModel, "id = ?", resourcePool.RegionId).Error; err != nil {
		return bizerr.WrapGormError(err, "failed to query region")
	}

	schemaResourcePool, err := enToModel(resourcePool)
	if err != nil {
		return err
	}
	dbErr := wdb.Create(schemaResourcePool).Error
	if dbErr != nil {
		return bizerr.WrapGormError(dbErr, "failed to create resource pool")
	}
	return nil
}

// Save updates an existing ResourcePool record in the database.
func (i *resourcePoolRepoImpl) Save(ctx context.Context, resourcePool *entity.ResourcePool) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.Save")
	defer span.End()

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

	schemaResourcePool, err := enToModel(resourcePool)
	if err != nil {
		return err
	}
	dbErr := wdb.Omit(clause.Associations, "created_at").Save(schemaResourcePool).Error
	if dbErr != nil {
		return bizerr.WrapGormError(dbErr, "failed to save resource pool")
	}

	return nil
}

// Load retrieves a ResourcePool record by its ID.
// If showDeleted is true, it will include soft-deleted records.
func (i *resourcePoolRepoImpl) Load(ctx context.Context, id uint64, showDeleted bool) (*entity.ResourcePool, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.Load")
	defer span.End()

	return i.load(ctx, id, showDeleted, false)
}

func (i *resourcePoolRepoImpl) LoadForUpdate(
	ctx context.Context,
	id uint64,
	showDeleted bool,
) (*entity.ResourcePool, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.LoadForUpdate")
	defer span.End()

	return i.load(ctx, id, showDeleted, true)
}

func (i *resourcePoolRepoImpl) LoadLifecycleForUpdate(
	ctx context.Context,
	id uint64,
	showDeleted bool,
) (*resourcepoolrepo.Lifecycle, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.LoadLifecycleForUpdate")
	defer span.End()
	if id == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "resource pool id is empty")
	}

	var row struct {
		LifecycleState string
		DeletedAt      gorm.DeletedAt
	}
	rdb := i.provider.WriteDB(ctx)
	if showDeleted {
		rdb = rdb.Unscoped()
	}
	if err := rdb.Model(&schema.ResourcePool{}).
		Clauses(clause.Locking{Strength: "UPDATE"}).
		Select("lifecycle_state", "deleted_at").
		Where("`resource_pools`.`id` = ?", id).
		Take(&row).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "failed to lock resource pool lifecycle")
	}

	state, err := lifecycleStateFromModel(row.LifecycleState)
	if err != nil {
		return nil, err
	}
	return &resourcepoolrepo.Lifecycle{
		State:     state,
		IsDeleted: row.DeletedAt.Valid,
	}, nil
}

func (i *resourcePoolRepoImpl) ListDueAutomaticResume(ctx context.Context) ([]uint64, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.ListDueAutomaticResume")
	defer span.End()

	ids := []uint64{}
	result := i.provider.ReadDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("lifecycle_state = ?", globalngv1.ResourcePool_PAUSED.String()).
		Where("automatic_resume_at <= CURRENT_TIMESTAMP(3)").
		Order("id").
		Pluck("id", &ids)
	if result.Error != nil {
		return nil, bizerr.WrapGormError(result.Error, "list resource pools due for automatic resume")
	}
	return ids, nil
}

func (i *resourcePoolRepoImpl) load(
	ctx context.Context,
	id uint64,
	showDeleted bool,
	forUpdate bool,
) (*entity.ResourcePool, bizerr.BizError) {
	if id == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "resource pool id is empty")
	}

	var schemaResourcePool schema.ResourcePool
	// Start every terminal query from the provider handle. Reusing a GORM query
	// after Unscoped and Take would carry the ResourcePool statement into the
	// Provider and Region lookups.
	newQuery := func() *gorm.DB {
		var query *gorm.DB
		if forUpdate {
			query = i.provider.WriteDB(ctx)
		} else {
			query = i.provider.ReadDB(ctx)
		}
		if showDeleted {
			query = query.Unscoped()
		}
		return query
	}

	if forUpdate {
		if err := newQuery().Model(&schema.ResourcePool{}).
			Clauses(clause.Locking{Strength: "UPDATE"}).
			Where("`resource_pools`.`id` = ?", id).
			Take(&schemaResourcePool).Error; err != nil {
			return nil, bizerr.WrapGormError(err, "failed to lock resource pool")
		}

		var provider schema.Provider
		if err := newQuery().Take(&provider, "id = ?", schemaResourcePool.ProviderId).Error; err != nil {
			return nil, bizerr.WrapGormError(err, "failed to get resource pool provider")
		}
		var region schema.Region
		if err := newQuery().Take(&region, "id = ?", schemaResourcePool.RegionId).Error; err != nil {
			return nil, bizerr.WrapGormError(err, "failed to get resource pool region")
		}
		schemaResourcePool.Provider = &provider
		schemaResourcePool.Region = &region
	} else {
		if err := newQuery().Model(&schema.ResourcePool{}).InnerJoins("Provider").InnerJoins("Region").
			Where("`resource_pools`.`id` = ?", id).
			Take(&schemaResourcePool).Error; err != nil {
			return nil, bizerr.WrapGormError(err, "failed to get resource pool")
		}
	}

	return modelToEn(&schemaResourcePool)
}

// Delete removes a ResourcePool record from the database by its ID.
func (i *resourcePoolRepoImpl) Delete(ctx context.Context, id uint64) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.Delete")
	defer span.End()

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

	wdb := i.provider.WriteDB(ctx)

	if err := wdb.Where("`id` = ?", id).Delete(&schema.ResourcePool{}).Error; err != nil {
		return bizerr.WrapGormError(err, "delete resource pool")
	}
	return nil
}

func (i *resourcePoolRepoImpl) LockOrganizationNamespace(
	ctx context.Context,
	organizationID uint64,
) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.LockOrganizationNamespace")
	defer span.End()

	if organizationID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "organization id is required")
	}

	wdb := i.provider.WriteDB(ctx)
	namespace := &schema.ResourcePoolNamespace{OrganizationId: organizationID}
	if err := wdb.Clauses(clause.OnConflict{DoNothing: true}).Create(namespace).Error; err != nil {
		return bizerr.WrapGormError(err, "create resource pool namespace")
	}
	if err := wdb.Clauses(clause.Locking{Strength: "UPDATE"}).
		Where("organization_id = ?", organizationID).
		Take(namespace).Error; err != nil {
		return bizerr.WrapGormError(err, "lock resource pool namespace")
	}

	return nil
}

func (i *resourcePoolRepoImpl) HasLiveNormalizedName(
	ctx context.Context,
	organizationID uint64,
	normalizedName string,
	excludeResourcePoolID uint64,
) (bool, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.HasLiveNormalizedName")
	defer span.End()

	if organizationID == 0 {
		return false, bizerr.New(bizerr.ErrCodeParams, "organization id is required")
	}
	if normalizedName == "" {
		return false, bizerr.New(bizerr.ErrCodeParams, "normalized name is required")
	}

	// Names are unique across all service plans within the organization.
	tx := i.liveOrganizationResourcePools(ctx, organizationID).
		Select("id", "name", "normalized_name").
		Where("normalized_name = ? OR normalized_name = ''", normalizedName)
	if excludeResourcePoolID != 0 {
		tx = tx.Where("id <> ?", excludeResourcePoolID)
	}

	var candidates []schema.ResourcePool
	if err := tx.Find(&candidates).Error; err != nil {
		return false, bizerr.WrapGormError(err, "check live resource pool name")
	}
	for i := range candidates {
		if candidates[i].NormalizedName == normalizedName ||
			entity.NormalizeResourcePoolName(candidates[i].Name) == normalizedName {
			return true, nil
		}
	}
	return false, nil
}

func (i *resourcePoolRepoImpl) HasLiveCIDROverlap(
	ctx context.Context,
	organizationID uint64,
	cloudProvider entity.CloudProviderName,
	region entity.RegionName,
	cidrStart []byte,
	cidrEnd []byte,
	excludeResourcePoolID uint64,
) (bool, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.HasLiveCIDROverlap")
	defer span.End()

	if organizationID == 0 {
		return false, bizerr.New(bizerr.ErrCodeParams, "organization id is required")
	}
	if cloudProvider == globalcommonv1.CloudProvider_NAME_UNSPECIFIED ||
		region == globalcommonv1.Region_NAME_UNSPECIFIED {
		return false, bizerr.New(bizerr.ErrCodeParams, "cloud provider and region are required")
	}
	if len(cidrStart) != 4 || len(cidrEnd) != 4 || bytes.Compare(cidrStart, cidrEnd) > 0 {
		return false, bizerr.New(bizerr.ErrCodeParams, "invalid canonical ipv4 range")
	}

	tx := i.liveBYOCOrganizationResourcePoolsInRegion(ctx, organizationID, cloudProvider, region).
		Select(
			"resource_pools.id",
			"resource_pools.cidr",
			"resource_pools.cidr_start_ipv4",
			"resource_pools.cidr_end_ipv4",
		).
		Where(
			"(cidr_start_ipv4 <= ? AND cidr_end_ipv4 >= ?) OR cidr_start_ipv4 IS NULL OR cidr_end_ipv4 IS NULL",
			cidrEnd,
			cidrStart,
		)
	if excludeResourcePoolID != 0 {
		tx = tx.Where("resource_pools.id <> ?", excludeResourcePoolID)
	}

	var candidates []schema.ResourcePool
	if err := tx.Find(&candidates).Error; err != nil {
		return false, bizerr.WrapGormError(err, "check live resource pool CIDR overlap")
	}
	for i := range candidates {
		candidateStart := candidates[i].CidrStartIPv4
		candidateEnd := candidates[i].CidrEndIPv4
		if len(candidateStart) != 4 || len(candidateEnd) != 4 {
			var err error
			candidateStart, candidateEnd, err = entity.CanonicalIPv4Range(candidates[i].Cidr)
			if err != nil {
				return false, bizerr.WrapErrorWithCode(
					err,
					bizerr.ErrCodeParams,
					"invalid legacy resource pool CIDR for resource pool %d",
					candidates[i].ID,
				)
			}
		}
		if bytes.Compare(candidateStart, cidrEnd) <= 0 && bytes.Compare(candidateEnd, cidrStart) >= 0 {
			return true, nil
		}
	}
	return false, nil
}

func (i *resourcePoolRepoImpl) ListLiveBYOCOrganizationResourcePoolsInRegion(
	ctx context.Context,
	organizationID uint64,
	cloudProvider entity.CloudProviderName,
	region entity.RegionName,
) ([]*entity.ResourcePool, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.ListLiveBYOCOrganizationResourcePoolsInRegion")
	defer span.End()

	if organizationID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "organization id is required")
	}
	if cloudProvider == globalcommonv1.CloudProvider_NAME_UNSPECIFIED ||
		region == globalcommonv1.Region_NAME_UNSPECIFIED {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cloud provider and region are required")
	}

	var models []*schema.ResourcePool
	if err := i.liveBYOCOrganizationResourcePoolsInRegion(ctx, organizationID, cloudProvider, region).
		Order("resource_pools.id ASC").
		Find(&models).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "list live BYOC organization resource pools in region")
	}
	if len(models) == 0 {
		return []*entity.ResourcePool{}, nil
	}

	wdb := i.provider.WriteDB(ctx)
	providerIDs := make([]uint64, 0, len(models))
	regionIDs := make([]uint64, 0, len(models))
	for _, model := range models {
		providerIDs = append(providerIDs, model.ProviderId)
		regionIDs = append(regionIDs, model.RegionId)
	}

	var providers []schema.Provider
	if err := wdb.Unscoped().Where("id IN ?", providerIDs).Find(&providers).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "load resource pool providers")
	}
	providerByID := make(map[uint64]*schema.Provider, len(providers))
	for idx := range providers {
		providerByID[uint64(providers[idx].ID)] = &providers[idx]
	}

	var regions []schema.Region
	if err := wdb.Unscoped().Where("id IN ?", regionIDs).Find(&regions).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "load resource pool regions")
	}
	regionByID := make(map[uint64]*schema.Region, len(regions))
	for idx := range regions {
		regionByID[uint64(regions[idx].ID)] = &regions[idx]
	}

	result := make([]*entity.ResourcePool, 0, len(models))
	for _, model := range models {
		provider, providerOK := providerByID[model.ProviderId]
		region, regionOK := regionByID[model.RegionId]
		if !providerOK || !regionOK {
			return nil, bizerr.New(
				bizerr.ErrCodeUnknown,
				"resource pool %d has invalid provider or region association",
				model.ID,
			)
		}
		model.Provider = provider
		model.Region = region
		resourcePool, err := modelToEn(model)
		if err != nil {
			return nil, err
		}
		result = append(result, resourcePool)
	}
	return result, nil
}

func (i *resourcePoolRepoImpl) InspectLiveCIDROverlapForBackfill(
	ctx context.Context,
	organizationID uint64,
	cloudProvider entity.CloudProviderName,
	region entity.RegionName,
	cidrStart []byte,
	cidrEnd []byte,
	excludeResourcePoolID uint64,
) (bool, []uint64, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.InspectLiveCIDROverlapForBackfill")
	defer span.End()

	if organizationID == 0 {
		return false, nil, bizerr.New(bizerr.ErrCodeParams, "organization id is required")
	}
	if cloudProvider == globalcommonv1.CloudProvider_NAME_UNSPECIFIED ||
		region == globalcommonv1.Region_NAME_UNSPECIFIED {
		return false, nil, bizerr.New(bizerr.ErrCodeParams, "cloud provider and region are required")
	}
	if len(cidrStart) != 4 || len(cidrEnd) != 4 || bytes.Compare(cidrStart, cidrEnd) > 0 {
		return false, nil, bizerr.New(bizerr.ErrCodeParams, "invalid canonical ipv4 range")
	}

	tx := i.liveBYOCOrganizationResourcePoolsInRegion(ctx, organizationID, cloudProvider, region).
		Select(
			"resource_pools.id",
			"resource_pools.cidr",
			"resource_pools.cidr_start_ipv4",
			"resource_pools.cidr_end_ipv4",
		).
		Where(
			"(cidr_start_ipv4 <= ? AND cidr_end_ipv4 >= ?) OR cidr_start_ipv4 IS NULL OR cidr_end_ipv4 IS NULL",
			cidrEnd,
			cidrStart,
		)
	if excludeResourcePoolID != 0 {
		tx = tx.Where("resource_pools.id <> ?", excludeResourcePoolID)
	}

	var candidates []schema.ResourcePool
	if err := tx.Find(&candidates).Error; err != nil {
		return false, nil, bizerr.WrapGormError(err, "inspect live resource pool CIDR overlap")
	}

	overlap := false
	invalidResourcePoolIDs := make([]uint64, 0)
	for i := range candidates {
		candidateStart := candidates[i].CidrStartIPv4
		candidateEnd := candidates[i].CidrEndIPv4
		if len(candidateStart) != 4 || len(candidateEnd) != 4 {
			var err error
			candidateStart, candidateEnd, err = entity.CanonicalIPv4Range(candidates[i].Cidr)
			if err != nil {
				invalidResourcePoolIDs = append(invalidResourcePoolIDs, candidates[i].ID)
				continue
			}
		}
		if bytes.Compare(candidateStart, cidrEnd) <= 0 && bytes.Compare(candidateEnd, cidrStart) >= 0 {
			overlap = true
		}
	}
	return overlap, invalidResourcePoolIDs, nil
}

func (i *resourcePoolRepoImpl) BackfillMetadata(
	ctx context.Context,
	patch *resourcepoolrepo.MetadataBackfillPatch,
) (resourcepoolrepo.MetadataBackfillResult, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.BackfillMetadata")
	defer span.End()

	var result resourcepoolrepo.MetadataBackfillResult
	if patch == nil || patch.ResourcePoolID == 0 {
		return result, bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	resourcePool := func() *gorm.DB {
		return i.provider.WriteDB(ctx).
			Model(&schema.ResourcePool{}).
			Where("id = ? AND deleted_at IS NULL", patch.ResourcePoolID)
	}

	if patch.NormalizedName != nil {
		tx := resourcePool().
			Where("normalized_name = ''").
			Update("normalized_name", *patch.NormalizedName)
		if tx.Error != nil {
			return result, bizerr.WrapGormError(tx.Error, "backfill resource pool normalized name")
		}
		result.NormalizedName = tx.RowsAffected > 0
	}
	if len(patch.CIDRStartIPv4) > 0 || len(patch.CIDREndIPv4) > 0 {
		if len(patch.CIDRStartIPv4) != 4 || len(patch.CIDREndIPv4) != 4 {
			return result, bizerr.New(bizerr.ErrCodeParams, "canonical resource pool CIDR is invalid")
		}
		tx := resourcePool().
			Where("cidr_start_ipv4 IS NULL OR cidr_end_ipv4 IS NULL").
			Updates(map[string]any{
				"cidr_start_ipv4": patch.CIDRStartIPv4,
				"cidr_end_ipv4":   patch.CIDREndIPv4,
			})
		if tx.Error != nil {
			return result, bizerr.WrapGormError(tx.Error, "backfill resource pool CIDR")
		}
		result.CIDR = tx.RowsAffected > 0
	}
	if patch.LifecycleState != nil {
		tx := resourcePool().
			Where("lifecycle_state = ''").
			Update("lifecycle_state", patch.LifecycleState.String())
		if tx.Error != nil {
			return result, bizerr.WrapGormError(tx.Error, "backfill resource pool lifecycle")
		}
		result.LifecycleState = tx.RowsAffected > 0
	}
	if patch.VcpuLimitMode != nil {
		tx := resourcePool().
			Where(
				"vcpu_limit_mode IN ('', ?) AND vcpus_limit > 0",
				globalngv1.ResourcePool_INHERITED.String(),
			).
			Update("vcpu_limit_mode", patch.VcpuLimitMode.String())
		if tx.Error != nil {
			return result, bizerr.WrapGormError(tx.Error, "backfill resource pool vCPU mode")
		}
		result.VcpuLimitMode = tx.RowsAffected > 0
	}
	if patch.BackfillTags {
		model := &schema.ResourcePool{CloudResourceTags: maps.Clone(patch.CloudResourceTags)}
		tx := resourcePool().
			Where("cloud_resource_tags_json IS NULL").
			Select("CloudResourceTags").
			Updates(model)
		if tx.Error != nil {
			return result, bizerr.WrapGormError(tx.Error, "backfill resource pool cloud resource tags")
		}
		result.CloudResourceTags = tx.RowsAffected > 0
	}

	return result, nil
}

func (i *resourcePoolRepoImpl) liveBYOCOrganizationResourcePools(
	ctx context.Context,
	organizationID uint64,
) *gorm.DB {
	return i.liveOrganizationResourcePools(ctx, organizationID).
		Where("resource_pools.service_plan = ?", globalcommonv1.ServicePlan_BYOC.String())
}

func (i *resourcePoolRepoImpl) liveOrganizationResourcePools(
	ctx context.Context,
	organizationID uint64,
) *gorm.DB {
	return i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where(
			"resource_pools.owner_type = ? AND resource_pools.owner_id = ?",
			globalngv1.ResourcePool_ORGANIZATION,
			organizationID,
		)
}

func (i *resourcePoolRepoImpl) liveBYOCOrganizationResourcePoolsInRegion(
	ctx context.Context,
	organizationID uint64,
	cloudProvider entity.CloudProviderName,
	region entity.RegionName,
) *gorm.DB {
	wdb := i.provider.WriteDB(ctx)
	providerIDs := wdb.Unscoped().
		Model(&schema.Provider{}).
		Select("id").
		Where("name = ?", cloudProvider.String())
	regionIDs := wdb.Unscoped().
		Model(&schema.Region{}).
		Select("id").
		Where("region_name = ?", region.String())

	return i.liveBYOCOrganizationResourcePools(ctx, organizationID).
		Where("resource_pools.provider_id IN (?)", providerIDs).
		Where("resource_pools.region_id IN (?)", regionIDs)
}

// Query retrieves a list of ResourcePool records matching the provided filter criteria.
func (i *resourcePoolRepoImpl) Query(
	ctx context.Context,
	filter *resourcepoolrepo.Filter,
) ([]*entity.ResourcePool, int64, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "resourcePoolRepo.Query")
	defer span.End()

	if filter == nil {
		return nil, 0, bizerr.New(bizerr.ErrCodeUnknown, "filter is nil")
	}

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

	var schemaResourcePools []*schema.ResourcePool
	tx := rdb.Model(&schema.ResourcePool{}).InnerJoins("Provider").InnerJoins("Region")

	if filter.ShowDeleted {
		tx = tx.Unscoped()
	}
	if len(filter.CloudProviderNames) > 0 {
		cloudProviders := lo.Map(
			filter.CloudProviderNames,
			func(cloudProvider entity.CloudProviderName, index int) string {
				return cloudProvider.String()
			},
		)
		tx = tx.Where("`Provider`.`name` IN ?", cloudProviders)
	}
	if len(filter.RegionNames) > 0 {
		tx = tx.Where(
			"`Region`.`region_name` IN ?",
			lo.Map(filter.RegionNames, func(region entity.RegionName, index int) string {
				return region.String()
			}),
		)
	}

	if filter.RegionId > 0 {
		tx = tx.Where(
			"`resource_pools`.`region_id` = ?", filter.RegionId,
		)
	}

	if len(filter.Ids) > 0 {
		tx = tx.Where("`resource_pools`.`id` IN ?", filter.Ids)
	}

	if filter.Name != "" {
		pattern := "%" + filter.Name + "%"
		tx = tx.Where(
			"`resource_pools`.`normalized_name` LIKE ? OR "+
				"(`resource_pools`.`normalized_name` = '' AND `resource_pools`.`name` LIKE ?)",
			pattern,
			pattern,
		)
	}

	if filter.OwnerType != 0 {
		tx = tx.Where("`owner_type` = ?", filter.OwnerType)
	}

	if len(filter.OwnerIds) > 0 {
		tx = tx.Where("`owner_id` IN ?", filter.OwnerIds)
	}

	if len(filter.ServicePlans) > 0 {
		servicePlans := lo.FilterMap(
			filter.ServicePlans,
			func(servicePlan entity.ServicePlan, _ int) (string, bool) {
				if servicePlan == globalcommonv1.ServicePlan_SERVICE_PLAN_UNSPECIFIED {
					return "", false
				}
				return servicePlan.String(), true
			},
		)
		if len(servicePlans) > 0 {
			tx = tx.Where("`resource_pools`.`service_plan` IN ?", servicePlans)
		}
	}
	if len(filter.HighAvailabilityTypes) > 0 {
		highAvailabilityTypes := make([]string, 0, len(filter.HighAvailabilityTypes)+1)
		for _, highAvailabilityType := range filter.HighAvailabilityTypes {
			// Unspecified pools can be stored as either the enum name or the schema default.
			highAvailabilityTypes = append(highAvailabilityTypes, highAvailabilityType.String())
			if highAvailabilityType == globalngv1.ResourcePool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED {
				highAvailabilityTypes = append(highAvailabilityTypes, "")
			}
		}
		tx = tx.Where("`resource_pools`.`high_availability_type` IN ?", highAvailabilityTypes)
	}
	if len(filter.LifecycleStates) > 0 {
		lifecycleStates := lo.Map(
			filter.LifecycleStates,
			func(lifecycleState globalngv1.ResourcePool_LifecycleState, _ int) string {
				return lifecycleState.String()
			},
		)
		tx = tx.Where("`resource_pools`.`lifecycle_state` IN ?", lifecycleStates)
	}

	if !filter.CreatedAfter.IsZero() {
		tx = tx.Where("`resource_pools`.`created_at` >= ?", filter.CreatedAfter)
	}
	if !filter.CreatedBefore.IsZero() {
		tx = tx.Where("`resource_pools`.`created_at` <= ?", filter.CreatedBefore)
	}

	var total int64
	if err := tx.Session(&gorm.Session{}).Count(&total).Error; err != nil {
		return nil, 0, bizerr.WrapGormError(err, "count resource pools")
	}

	if filter.Cursor != nil {
		tx = tx.Where(
			"(`resource_pools`.`created_at` < ?) OR (`resource_pools`.`created_at` = ? AND `resource_pools`.`id` < ?)",
			filter.Cursor.CreatedAt,
			filter.Cursor.CreatedAt,
			filter.Cursor.ID,
		)
	}

	tx = tx.Order("`resource_pools`.`created_at` DESC").Order("`resource_pools`.`id` DESC")

	if filter.PageSize > 0 {
		tx = tx.Limit(int(filter.PageSize))
		if filter.Cursor == nil && filter.Offset > 0 {
			tx = tx.Offset(int(filter.Offset))
		}
	}

	if err := tx.Find(&schemaResourcePools).Error; err != nil {
		return nil, 0, bizerr.WrapGormError(err, "query resource pool")
	}

	var resourcePools []*entity.ResourcePool
	for _, schemaResourcePool := range schemaResourcePools {
		en, err := modelToEn(schemaResourcePool)
		if err != nil {
			return nil, 0, err
		}
		resourcePools = append(resourcePools, en)
	}

	return resourcePools, total, nil
}

func (i *resourcePoolRepoImpl) UpdateDataplaneId(ctx context.Context,
	id uint64,
	dataplaneId string) bizerr.BizError {
	if id == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if dataplaneId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "dataplane id is required")
	}

	wdb := i.provider.WriteDB(ctx)
	err := wdb.Transaction(func(tx *gorm.DB) error {
		var resourcePool schema.ResourcePool
		err := tx.Set("gorm:query_option", "FOR UPDATE").
			Where("id = ?", id).
			First(&resourcePool).Error

		if err != nil {
			return bizerr.WrapGormError(err, "failed to load resource pool for update")
		}

		if resourcePool.DataplaneId == dataplaneId {
			return nil
		}

		if resourcePool.DataplaneId != "" {
			return bizerr.New(bizerr.ErrCodeParams, "cannot modify existed dataplane id")
		}

		updateErr := tx.Model(&resourcePool).Update("dataplane_id", dataplaneId).Error
		if updateErr != nil {
			return updateErr
		}

		return nil
	})
	if err != nil {
		return bizerr.WrapGormError(err, "update dataplane failed")
	}
	return nil
}

func (i *resourcePoolRepoImpl) UpdateLifecycle(
	ctx context.Context,
	id uint64,
	state globalngv1.ResourcePool_LifecycleState,
	messageCode string,
) bizerr.BizError {
	if id == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if state == globalngv1.ResourcePool_LIFECYCLE_STATE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool lifecycle state is required")
	}

	model := &schema.ResourcePool{
		LifecycleState:   state.String(),
		StateMessageCode: messageCode,
	}
	result := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("id = ?", id).
		Select("LifecycleState", "StateMessageCode").
		Updates(model)
	return i.validateSingleResourcePoolUpdate(ctx, id, result, "update resource pool lifecycle")
}

func (i *resourcePoolRepoImpl) CompletePauseResumeLifecycle(
	ctx context.Context,
	id uint64,
	expectedState globalngv1.ResourcePool_LifecycleState,
	targetState globalngv1.ResourcePool_LifecycleState,
) (bool, bizerr.BizError) {
	if id == 0 {
		return false, bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if expectedState == globalngv1.ResourcePool_LIFECYCLE_STATE_UNSPECIFIED ||
		targetState == globalngv1.ResourcePool_LIFECYCLE_STATE_UNSPECIFIED {
		return false, bizerr.New(bizerr.ErrCodeParams, "resource pool lifecycle state is required")
	}

	updates := map[string]any{
		"lifecycle_state":     targetState.String(),
		"state_message_code":  "",
		"automatic_resume_at": nil,
	}
	if targetState == globalngv1.ResourcePool_PAUSED {
		updates["paused_at"] = gorm.Expr("CURRENT_TIMESTAMP(3)")
		updates["automatic_resume_at"] = gorm.Expr("DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL 7 DAY)")
	} else {
		updates["paused_at"] = nil
	}

	result := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("id = ? AND lifecycle_state = ?", id, expectedState.String()).
		Updates(updates)
	if result.Error != nil {
		return false, bizerr.WrapGormError(result.Error, "complete resource pool pause/resume lifecycle")
	}
	if result.RowsAffected == 1 {
		return true, nil
	}

	// A zero-row conditional update is ambiguous: another worker may have
	// completed the transition, or the pool may have been soft-deleted after
	// the workflow loaded it. Re-read both states so these benign races remain
	// successful; only a different live lifecycle state means ownership loss.
	var current struct {
		LifecycleState string
		DeletedAt      gorm.DeletedAt
	}
	readResult := i.provider.WriteDB(ctx).
		Unscoped().
		Model(&schema.ResourcePool{}).
		Select("lifecycle_state", "deleted_at").
		Where("id = ?", id).
		First(&current)
	if errors.Is(readResult.Error, gorm.ErrRecordNotFound) {
		return false, nil
	}
	if readResult.Error != nil {
		return false, bizerr.WrapGormError(readResult.Error, "re-read resource pool pause/resume lifecycle")
	}
	if current.DeletedAt.Valid || current.LifecycleState == targetState.String() {
		return true, nil
	}
	return false, nil
}

func (i *resourcePoolRepoImpl) CompareAndSetLifecycle(
	ctx context.Context,
	id uint64,
	expectedState globalngv1.ResourcePool_LifecycleState,
	expectedMessageCode string,
	expectedUpdatedAt time.Time,
	state globalngv1.ResourcePool_LifecycleState,
	messageCode string,
) (bool, bizerr.BizError) {
	if id == 0 {
		return false, bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if state == globalngv1.ResourcePool_LIFECYCLE_STATE_UNSPECIFIED {
		return false, bizerr.New(bizerr.ErrCodeParams, "resource pool lifecycle state is required")
	}

	expectedStateValue := ""
	if expectedState != globalngv1.ResourcePool_LIFECYCLE_STATE_UNSPECIFIED {
		expectedStateValue = expectedState.String()
	}
	model := &schema.ResourcePool{
		LifecycleState:   state.String(),
		StateMessageCode: messageCode,
	}
	result := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where(
			"id = ? AND lifecycle_state = ? AND state_message_code = ? AND updated_at = ?",
			id,
			expectedStateValue,
			expectedMessageCode,
			expectedUpdatedAt,
		).
		Select("LifecycleState", "StateMessageCode").
		Updates(model)
	if result.Error != nil {
		return false, bizerr.WrapGormError(result.Error, "compare and set resource pool lifecycle")
	}
	return result.RowsAffected == 1, nil
}

func (i *resourcePoolRepoImpl) UpdateInstanceLimit(
	ctx context.Context,
	id uint64,
	limit uint64,
) bizerr.BizError {
	if id == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if limit == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool instance limit must be positive")
	}
	result := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("id = ?", id).
		Update("instance_limit", limit)
	return i.validateSingleResourcePoolUpdate(ctx, id, result, "update resource pool instance limit")
}

func (i *resourcePoolRepoImpl) UpdateVcpuPolicy(
	ctx context.Context,
	id uint64,
	mode globalngv1.ResourcePool_VcpuLimitMode,
	limit uint64,
) bizerr.BizError {
	if id == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}
	if mode == globalngv1.ResourcePool_VCPU_LIMIT_MODE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool vCPU limit mode is required")
	}

	model := &schema.ResourcePool{
		VcpuLimitMode: mode.String(),
		VcpusLimit:    limit,
	}
	result := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("id = ?", id).
		Select("VcpuLimitMode", "VcpusLimit").
		Updates(model)
	return i.validateSingleResourcePoolUpdate(ctx, id, result, "update resource pool vCPU policy")
}

func (i *resourcePoolRepoImpl) UpdateCloudResourceTags(
	ctx context.Context,
	id uint64,
	tags map[string]string,
) bizerr.BizError {
	if id == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "resource pool id is required")
	}

	model := &schema.ResourcePool{CloudResourceTags: maps.Clone(tags)}
	result := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("id = ?", id).
		Select("CloudResourceTags").
		Updates(model)
	return i.validateSingleResourcePoolUpdate(ctx, id, result, "update resource pool cloud resource tags")
}

func (i *resourcePoolRepoImpl) validateSingleResourcePoolUpdate(
	ctx context.Context,
	id uint64,
	result *gorm.DB,
	operation string,
) bizerr.BizError {
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "%s", operation)
	}
	if result.RowsAffected == 1 {
		return nil
	}
	if result.RowsAffected > 1 {
		return bizerr.New(bizerr.ErrCodeUnknown, "%s affected more than one resource pool", operation)
	}

	var count int64
	if err := i.provider.WriteDB(ctx).
		Model(&schema.ResourcePool{}).
		Where("id = ?", id).
		Count(&count).Error; err != nil {
		return bizerr.WrapGormError(err, "%s existence check", operation)
	}
	if count == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "resource pool not found")
	}
	return nil
}