JCS-pub/datamap/internal/models/blocktransfer.go

package models

import (
	"errors"
	"log"
	"strconv"
	"time"

	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
	stgmod "gitlink.org.cn/cloudream/storage/common/models"
	"gitlink.org.cn/cloudream/storage/common/pkgs/sysevent"
	"gorm.io/gorm"
)

type StorageTransferCount struct {
	RelationshipID    int64     `gorm:"column:RelationshipID; primaryKey; type:bigint; autoIncrement" json:"relationshipID"`
	ObjectID          int64     `gorm:"column:ObjectID; type:bigint; not null" json:"objectID"`
	Status            int64     `gorm:"column:Status; type:bigint; not null" json:"status"`                       // 连线左侧的状态
	SourceStorageID   int64     `gorm:"column:SourceStorageID; type:bigint; not null" json:"sourceStorageID"`     // 源存储节点 ID
	TargetStorageID   int64     `gorm:"column:TargetStorageID; type:bigint; not null" json:"targetStorageID"`     // 目标存储节点 ID
	DataTransferCount int64     `gorm:"column:DataTransferCount; type:bigint; not null" json:"dataTransferCount"` // 数据传输量
	Timestamp         time.Time `gorm:"column:Timestamp; type:datatime; not null" json:"timestamp"`               // 变化结束时间戳
}

func (StorageTransferCount) TableName() string {
	return "storagetransfercount"
}

type StorageTransferCountRepository struct {
	repo *GormRepository
}

func NewStorageTransferCountRepository(db *gorm.DB) *StorageTransferCountRepository {
	return &StorageTransferCountRepository{repo: NewGormRepository(db)}
}

func (r *StorageTransferCountRepository) CreateStorageTransferCount(storageTransferCount *StorageTransferCount) error {
	return r.repo.Create(storageTransferCount)
}

func (r *StorageTransferCountRepository) UpdateStorageTransferCount(storageTransferCount *StorageTransferCount) error {
	return r.repo.Update(storageTransferCount)
}

func (r *StorageTransferCountRepository) GetStorageTransferCountByID(id int) (*StorageTransferCount, error) {
	var storageTransferCount StorageTransferCount
	err := r.repo.GetByID(uint(id), &storageTransferCount)
	if err != nil {
		return nil, err
	}
	return &storageTransferCount, nil
}

func (r *StorageTransferCountRepository) GetStorageTransferCountByObjectID(objectID int64) ([]StorageTransferCount, error) {
	var storageTransferCounts []StorageTransferCount
	query := "SELECT * FROM storagetransfercount WHERE ObjectID = ?"
	err := r.repo.db.Raw(query, objectID).Scan(&storageTransferCounts).Error
	if err != nil {
		return nil, err
	}
	return storageTransferCounts, nil
}

func (r *StorageTransferCountRepository) GetAllStorageTransferCounts() ([]StorageTransferCount, error) {
	var storageTransferCounts []StorageTransferCount
	err := r.repo.GetAll(&storageTransferCounts)
	if err != nil {
		return nil, err
	}
	return storageTransferCounts, nil
}

type BlockTransferWatcher struct {
	Name string
}

func (w *BlockTransferWatcher) OnEvent(event sysevent.SysEvent) {
	body, ok := event.Body.(*stgmod.BodyBlockTransfer)
	if !ok {
		return
	}

	repoDist := NewBlockDistributionRepository(DB)
	repoStorage := NewStorageRepository(DB)
	repoStorageTrans := NewStorageTransferCountRepository(DB)
	repoObject := NewObjectRepository(DB)

	for _, change := range body.BlockChanges {

		objectID, _ := strconv.ParseInt(string(body.ObjectID), 10, 64)
		object, _ := repoObject.GetObjectByID(objectID)
		// index, _ := strconv.ParseInt(change.Index, 10, 64)
		// sourceStorageID, _ := strconv.ParseInt(string(change.SourceStorageID), 10, 64)
		// targetStorageID, _ := strconv.ParseInt(string(change.TargetStorageID), 10, 64)
		// newDataCount, _ := strconv.ParseInt(change.DataTransferCount, 10, 64)

		switch change := change.(type) {
		case *stgmod.BlockChangeClone: //拷贝
			// TODO 从change中获取index, sourceStorageID, targetStorageID, newDataCount，下同

			//查询出存储在数据库中的BlockDistribution信息
			blockSource, errSource := repoDist.GetBlockDistributionByIndex(objectID, index, sourceStorageID)
			//没有记录就将source和target的信息都保存到库中
			if errors.Is(errSource, gorm.ErrRecordNotFound) {
				err := repoDist.CreateBlockDistribution(&BlockDistribution{
					ObjectID:  objectID,
					Type:      change.BlockType,
					Index:     index,
					StorageID: sourceStorageID,
					Status:    StatusNow,
					Timestamp: time.Now(),
				})
				if err != nil {
					log.Printf("Error create source blockdistribution: %v", err)
				}
			} else {
				//有数据则新增一条storageID为targetStorageID的记录,同时更新状态
				err := repoDist.CreateBlockDistribution(&BlockDistribution{
					ObjectID:  blockSource.ObjectID,
					Type:      change.BlockType,
					Index:     index,
					StorageID: targetStorageID,
					Status:    StatusNow,
					Timestamp: time.Now(),
				})
				if err != nil {
					log.Printf("Error update blockdistribution: %v", err)
				}
				//复制完成之后增加的dataCount要加到targetStorage的记录中
				storageOld, err := repoStorage.GetStorageByID(targetStorageID)
				if errors.Is(err, gorm.ErrRecordNotFound) {
					err = repoStorage.CreateStorage(&Storage{
						StorageID: cdssdk.StorageID(targetStorageID),
						DataCount: newDataCount,
						Timestamp: time.Now(),
					})
					if err != nil {
						log.Printf("Error increase datacount in targetstorage: %v", err)
					}
				} else {
					err = repoStorage.UpdateStorage(&Storage{
						StorageID: cdssdk.StorageID(targetStorageID),
						DataCount: storageOld.DataCount + newDataCount,
						Timestamp: time.Now(),
					})
					if err != nil {
						log.Printf("Error increase datacount in targetstorage: %v", err)
					}
				}

			}
			//新增记录到storageTransferCount表中
			err := repoStorageTrans.CreateStorageTransferCount(&StorageTransferCount{
				ObjectID:          objectID,
				Status:            int64(blockSource.Status),
				SourceStorageID:   sourceStorageID,
				TargetStorageID:   targetStorageID,
				DataTransferCount: newDataCount,
				Timestamp:         time.Now(),
			})
			if err != nil {
				log.Printf("Error create StorageTransferCount : %v", err)
			}
		case *stgmod.BlockChangeEnDecode: //编解码
			//删除所有的sourceBlock
			for _, sourceBlock := range change.SourceBlocks {
				sourceBlockIndex, _ := strconv.ParseInt(sourceBlock.Index, 10, 64)
				err := repoDist.DeleteBlockDistribution(objectID, sourceBlockIndex, sourceStorageID)
				if err != nil {
					log.Printf("Error delete blockdistribution: %v", err)
				}
			}
			//插入所有的targetBlock
			for _, targetBlock := range change.TargetBlocks {
				storageID, _ := strconv.ParseInt(string(targetBlock.StorageID), 10, 64)
				err := repoDist.CreateBlockDistribution(&BlockDistribution{
					ObjectID: objectID,
					Type:     targetBlock.BlockType,
					Index:    index,
					//直接保存到目标中心
					StorageID: storageID,
					Status:    StatusNow,
					Timestamp: time.Now(),
				})
				if err != nil {
					log.Printf("Error create blockdistribution: %v", err)
				}
			}
			//新增记录到storageTransferCount表中
			err := repoStorageTrans.CreateStorageTransferCount(&StorageTransferCount{
				ObjectID:          objectID,
				Status:            int64(object.Status),
				SourceStorageID:   sourceStorageID,
				TargetStorageID:   targetStorageID,
				DataTransferCount: newDataCount,
				Timestamp:         time.Now(),
			})
			if err != nil {
				log.Printf("Error create StorageTransferCount : %v", err)
			}

		case *stgmod.BlockChangeDeleted: //删除
			storageID, _ := strconv.ParseInt(string(change.StorageID), 10, 64)
			changeIndex, _ := strconv.ParseInt(change.Index, 10, 64)
			err := repoDist.DeleteBlockDistribution(objectID, changeIndex, storageID)
			if err != nil {
				log.Printf("Error delete blockdistribution: %v", err)
			}

		// case *stgmod.BlockChangeUpdated: //更新
		// 	for _, blockUpdate := range change.Blocks {
		// 		//查询出存储在数据库中的BlockDistribution信息
		// 		blockIndex, _ := strconv.ParseInt(blockUpdate.Index, 10, 64)
		// 		blockOld, err := repoDist.GetBlockDistributionByIndex(objectID, blockIndex, sourceStorageID)
		// 		newStorageID, _ := strconv.ParseInt(string(blockUpdate.StorageID), 10, 64)
		// 		err = repoDist.UpdateBlockDistribution(&BlockDistribution{
		// 			BlockID:   blockOld.BlockID,
		// 			ObjectID:  blockOld.ObjectID,
		// 			Type:      blockUpdate.BlockType,
		// 			Index:     blockIndex,
		// 			StorageID: newStorageID,
		// 			Status:    StatusNow,
		// 			Timestamp: time.Now(),
		// 		})
		// 		if err != nil {
		// 			log.Printf("Error delete blockdistribution: %v", err)
		// 		}
		// 	}

		default:
			break
		}
	}
}