pcm-coordinator/internal/logic/core/getclusterresourcespeclogic.go

package core

import (
	"context"
	"errors"
	"fmt"
	"github.com/mitchellh/mapstructure"
	"github.com/rs/zerolog/log"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/logic/schedule"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils"
	"time"

	"github.com/zeromicro/go-zero/core/logx"
	"gorm.io/gorm"
)

type GetClusterResourceSpecLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

const (
	ChangeTypeCreated  = 0
	ChangeTypeModified = 1
	ChangeTypeDeleted  = 2
)

type APIResponse struct {
	ClusterId   string     `json:"ClusterId"`
	ClusterType string     `json:"clusterType"`
	Region      string     `json:"region"`
	Resources   []Resource `json:"resources"`
	Msg         string     `json:"msg"`
}

type Resource struct {
	Resource      ResourceDetail   `json:"resource"`
	BaseResources []ResourceDetail `json:"baseResources"`
}

type ResourceDetail struct {
	Type      string `json:"type"`
	Name      string `json:"name"`
	Total     Metric `json:"total"`
	Available Metric `json:"available"`
}

type Metric struct {
	Unit  string  `json:"unit"`
	Value float64 `json:"value"`
}

func NewGetClusterResourceSpecLogic(ctx context.Context, svcCtx *svc.ServiceContext) *GetClusterResourceSpecLogic {
	return &GetClusterResourceSpecLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *GetClusterResourceSpecLogic) GetClusterResourceSpec(req *types.ResourceSpecReq) (*types.PageResult, error) {
	if req.ClusterId == "" {
		return nil, errors.New("ClusterId is required")
	}

	// 获取集群资源数据
	startTime := time.Now()
	apiResources, err := l.fetchClusterResources(req.ClusterId)
	log.Debug().Msgf("调用获取ai训练资源接口耗时： %v", time.Since(startTime))
	if err != nil {
		return nil, fmt.Errorf("failed to fetch cluster resources: %w", err)
	}

	// 同步资源到数据库
	if err := l.syncResourcesToDB(apiResources); err != nil {
		return nil, fmt.Errorf("failed to sync resources: %w", err)
	}

	// 查询数据库结果
	return l.queryDatabaseResults(req)
}

func (l *GetClusterResourceSpecLogic) fetchClusterResources(ClusterId string) ([]APIResponse, error) {
	queryLogic := schedule.NewQueryResourcesLogic(l.ctx, l.svcCtx)
	resources, err := queryLogic.QueryResources(&types.QueryResourcesReq{
		ClusterIDs: []string{ClusterId},
	})
	if err != nil {
		return nil, fmt.Errorf("query resources failed: %w", err)
	}

	var apiResponses []APIResponse
	if err := l.decodeAPIResponse(resources.Data, &apiResponses); err != nil {
		return nil, fmt.Errorf("decode response failed: %w", err)
	}

	return apiResponses, nil
}

func (l *GetClusterResourceSpecLogic) decodeAPIResponse(input interface{}, output *[]APIResponse) error {
	config := &mapstructure.DecoderConfig{
		Result:      output,
		TagName:     "json",
		ErrorUnused: true,
		DecodeHook: mapstructure.ComposeDecodeHookFunc(
			mapstructure.StringToTimeHookFunc(time.RFC3339),
			mapstructure.StringToSliceHookFunc(","),
		),
	}

	decoder, err := mapstructure.NewDecoder(config)
	if err != nil {
		return fmt.Errorf("failed to create decoder: %w", err)
	}

	if err := decoder.Decode(input); err != nil {
		return fmt.Errorf("decoding error: %w", err)
	}

	return nil
}

func (l *GetClusterResourceSpecLogic) syncResourcesToDB(apiResponses []APIResponse) error {
	for _, response := range apiResponses {
		// 转换API响应到数据库模型
		dbSpecs, apiSpecs, err := l.processAPIResponse(response)
		if err != nil {
			return err
		}

		// 处理资源变更
		if err := l.handleResourceChanges(dbSpecs, apiSpecs); err != nil {
			return fmt.Errorf("failed to handle resource changes: %w", err)
		}
	}
	return nil
}

func (l *GetClusterResourceSpecLogic) processAPIResponse(response APIResponse) ([]models.TResourceSpec, []models.TResourceSpec, error) {
	ClusterId := utils.StringToInt64(response.ClusterId)
	var dbSpecs []models.TResourceSpec
	if err := l.svcCtx.DbEngin.Model(models.TResourceSpec{}).Preload("BaseResourceSpecs").
		Where("cluster_id = ?", ClusterId).
		Find(&dbSpecs).Error; err != nil {
		return nil, nil, fmt.Errorf("database query failed: %w", err)
	}

	var apiSpecs []models.TResourceSpec
	for _, res := range response.Resources {
		spec := l.convertToResourceSpec(ClusterId, res)
		apiSpecs = append(apiSpecs, spec)
	}

	return dbSpecs, apiSpecs, nil
}

func (l *GetClusterResourceSpecLogic) handleResourceChanges(dbSpecs, apiSpecs []models.TResourceSpec) error {
	// 创建资源映射
	dbMap := make(map[string]models.TResourceSpec)
	for _, spec := range dbSpecs {
		key := resourceKey(spec.Type, spec.Name)
		dbMap[key] = spec
	}

	apiMap := make(map[string]models.TResourceSpec)
	for _, spec := range apiSpecs {
		key := resourceKey(spec.Type, spec.Name)
		apiMap[key] = spec
	}

	// 处理新增或更新的资源
	for key, apiSpec := range apiMap {
		dbSpec, exists := dbMap[key]
		if !exists {
			if err := l.createNewResource(&apiSpec); err != nil {
				return err
			}
			continue
		}

		if l.isSpecChanged(dbSpec, apiSpec) {
			if err := l.updateResource(&dbSpec, apiSpec); err != nil {
				return err
			}
		}
	}

	// 处理删除的资源
	for key, dbSpec := range dbMap {
		if _, exists := apiMap[key]; !exists {
			if err := l.markResourceDeleted(dbSpec.Id); err != nil {
				return err
			}
		}
	}

	return nil
}

func resourceKey(resType, name string) string {
	return fmt.Sprintf("%s::%s", resType, name)
}

func (l *GetClusterResourceSpecLogic) createNewResource(spec *models.TResourceSpec) error {
	return l.svcCtx.DbEngin.Transaction(func(tx *gorm.DB) error {
		if err := tx.Create(spec).Error; err != nil {
			return fmt.Errorf("failed to create resource: %w", err)
		}

		return nil
	})
}

func (l *GetClusterResourceSpecLogic) updateResource(existing *models.TResourceSpec, newSpec models.TResourceSpec) error {
	return l.svcCtx.DbEngin.Transaction(func(tx *gorm.DB) error {
		updates := map[string]interface{}{
			"total_count":     newSpec.TotalCount,
			"available_count": newSpec.AvailableCount,
			"change_type":     ChangeTypeModified,
			"update_time":     time.Now(),
		}

		if err := tx.Model(existing).Updates(updates).Error; err != nil {
			return fmt.Errorf("failed to update resource: %w", err)
		}

		return l.syncBaseResources(tx, existing.Id, newSpec.BaseResourceSpecs)
	})
}

func (l *GetClusterResourceSpecLogic) syncBaseResources(tx *gorm.DB, specID int64, newResources []models.TBaseResourceSpec) error {
	// 处理基础资源更新
	var existingResources []models.TBaseResourceSpec
	if err := tx.Where("resource_spec_id = ?", specID).Find(&existingResources).Error; err != nil {
		return fmt.Errorf("failed to query base resources: %w", err)
	}

	existingMap := make(map[string]models.TBaseResourceSpec)
	for _, r := range existingResources {
		key := resourceKey(r.Type, r.Name)
		existingMap[key] = r
	}

	// 处理更新和新增
	for i, newRes := range newResources {
		newRes.ResourceSpecId = specID
		key := resourceKey(newRes.Type, newRes.Name)

		if existing, exists := existingMap[key]; exists {
			newRes.Id = existing.Id
			newRes.CreateTime = existing.CreateTime
			if err := tx.Save(&newRes).Error; err != nil {
				return fmt.Errorf("failed to update base resource: %w", err)
			}
		} else {
			if err := tx.Create(&newRes).Error; err != nil {
				return fmt.Errorf("failed to create base resource: %w", err)
			}
		}
		newResources[i] = newRes
	}

	// 处理删除
	currentIDs := make(map[int64]struct{})
	for _, r := range newResources {
		currentIDs[r.Id] = struct{}{}
	}

	for _, existing := range existingResources {
		if _, exists := currentIDs[existing.Id]; !exists {
			if err := tx.Delete(&existing).Error; err != nil {
				return fmt.Errorf("failed to delete base resource: %w", err)
			}
		}
	}

	return nil
}

func (l *GetClusterResourceSpecLogic) markResourceDeleted(id int64) error {
	return l.svcCtx.DbEngin.Model(&models.TResourceSpec{}).
		Where("id = ?", id).
		Update("change_type", ChangeTypeDeleted).
		Error
}

func (l *GetClusterResourceSpecLogic) isSpecChanged(old, new models.TResourceSpec) bool {
	if old.TotalCount != new.TotalCount ||
		old.AvailableCount != new.AvailableCount ||
		old.Region != new.Region {
		return true
	}

	// 比较基础资源
	oldBaseMap := make(map[string]models.TBaseResourceSpec)
	for _, br := range old.BaseResourceSpecs {
		oldBaseMap[resourceKey(br.Type, br.Name)] = br
	}

	for _, newBr := range new.BaseResourceSpecs {
		key := resourceKey(newBr.Type, newBr.Name)
		oldBr, exists := oldBaseMap[key]
		if !exists ||
			oldBr.TotalValue != newBr.TotalValue ||
			oldBr.AvailableValue != newBr.AvailableValue {
			return true
		}
		delete(oldBaseMap, key)
	}

	return len(oldBaseMap) > 0
}

func (l *GetClusterResourceSpecLogic) convertToResourceSpec(ClusterId int64, res Resource) models.TResourceSpec {
	spec := models.TResourceSpec{
		Type:           res.Resource.Type,
		Name:           res.Resource.Name,
		TotalCount:     int64(res.Resource.Total.Value),
		AvailableCount: int64(res.Resource.Available.Value),
		ClusterId:      ClusterId,
		CreateTime:     time.Now(),
		UpdateTime:     time.Now(),
		ChangeType:     ChangeTypeCreated,
	}

	for _, br := range res.BaseResources {
		spec.BaseResourceSpecs = append(spec.BaseResourceSpecs, models.TBaseResourceSpec{
			Type:           br.Type,
			Name:           br.Name,
			TotalValue:     br.Total.Value,
			TotalUnit:      br.Total.Unit,
			AvailableValue: br.Available.Value,
			AvailableUnit:  br.Available.Unit,
			CreateTime:     time.Now(),
			UpdateTime:     time.Now(),
		})
	}

	return spec
}

func (l *GetClusterResourceSpecLogic) queryDatabaseResults(req *types.ResourceSpecReq) (*types.PageResult, error) {
	result := &types.PageResult{
		PageNum:  req.PageNum,
		PageSize: req.PageSize,
	}

	query := l.buildBaseQuery(req)
	if err := query.Count(&result.Total).Error; err != nil {
		return nil, fmt.Errorf("failed to count records: %w", err)
	}

	var specs []*models.TResourceSpec
	if err := query.Model(models.TResourceSpec{}).Preload("BaseResourceSpecs").
		Scopes(paginate(req.PageNum, req.PageSize)).
		Find(&specs).Error; err != nil {
		return nil, fmt.Errorf("failed to query resources: %w", err)
	}

	result.List = specs
	return result, nil
}

func (l *GetClusterResourceSpecLogic) buildBaseQuery(req *types.ResourceSpecReq) *gorm.DB {
	query := l.svcCtx.DbEngin.Model(&models.TResourceSpec{}).
		Where("cluster_id = ?", utils.StringToInt64(req.ClusterId)).
		Where("deleted_at IS NULL")

	if req.Status != "" {
		query = query.Where("status = ?", req.Status)
	}
	if req.ChangeType != "" {
		query = query.Where("change_type = ?", req.ChangeType)
	}
	if req.Type != "" {
		query = query.Where("type = ?", req.Type)
	}
	if req.Name != "" {
		query = query.Where("name LIKE ?", "%"+req.Name+"%")
	}

	return query
}

func paginate(pageNum, pageSize int) func(db *gorm.DB) *gorm.DB {
	return func(db *gorm.DB) *gorm.DB {
		offset := (pageNum - 1) * pageSize
		return db.Offset(offset).Limit(pageSize).Order("create_time DESC")
	}
}