1042 lines
28 KiB
Go
1042 lines
28 KiB
Go
package db
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"gitlink.org.cn/cloudream/common/pkgs/logger"
|
||
sch "gitlink.org.cn/cloudream/common/sdks/pcmscheduler"
|
||
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
|
||
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
|
||
uploadersdk "gitlink.org.cn/cloudream/common/sdks/uploader"
|
||
"gitlink.org.cn/cloudream/common/utils/serder"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
type UploadDataDB struct {
|
||
*DB
|
||
}
|
||
|
||
func (db *DB) UploadData() *UploadDataDB {
|
||
return &UploadDataDB{DB: db}
|
||
}
|
||
|
||
func (db *UploadDataDB) GetByPackageIDs(ctx SQLContext, packageIDs []cdssdk.PackageID, bindingIDs []int64) ([]uploadersdk.PackageDAO, error) {
|
||
var ret []uploadersdk.PackageDAO
|
||
err := ctx.Table("package").Where("package_id IN ? or binding_id IN ?", packageIDs, bindingIDs).
|
||
Preload("UploadedCluster").Find(&ret).Error
|
||
return ret, err
|
||
}
|
||
|
||
func (db *UploadDataDB) GetByPackageID(ctx SQLContext, packageID cdssdk.PackageID) (uploadersdk.PackageDAO, error) {
|
||
var ret uploadersdk.PackageDAO
|
||
err := ctx.Table("package").Where("package_id = ?", packageID).Preload("UploadedCluster").Find(&ret).Error
|
||
return ret, err
|
||
}
|
||
|
||
func (db *UploadDataDB) GetPackageByDataType(ctx SQLContext, userID cdssdk.UserID, dataType string) ([]uploadersdk.PackageDAO, error) {
|
||
var ret []uploadersdk.PackageDAO
|
||
err := ctx.Table("package").Where("user_id = ? and data_type = ?", userID, dataType).
|
||
Preload("UploadedCluster").Find(&ret).Error
|
||
return ret, err
|
||
}
|
||
|
||
func (db *UploadDataDB) GetByID(ctx SQLContext, ids []uploadersdk.DataID) ([]uploadersdk.Package, error) {
|
||
var ret []uploadersdk.Package
|
||
err := ctx.Table("UploadData").Where("ID IN ?", ids).Find(&ret).Error
|
||
return ret, err
|
||
}
|
||
|
||
func (db *UploadDataDB) QueryFolder(ctx SQLContext, queryParams sch.QueryData) ([]uploadersdk.Folder, error) {
|
||
var ret []uploadersdk.Folder
|
||
|
||
// 构建查询条件
|
||
pathName := sch.Split + queryParams.Path + "%"
|
||
err := ctx.Table("folders").Where("package_id = ? and path_name like ?", queryParams.PackageID, pathName).Find(&ret).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 使用 map 来去重路径
|
||
pathSet := make(map[string]time.Time)
|
||
|
||
// 计算 queryPath 和 stepNum 只需要一次
|
||
queryPath := sch.Split + queryParams.Path + sch.Split
|
||
stepNum := 2
|
||
if queryParams.Path == "" {
|
||
queryPath = sch.Split
|
||
stepNum = 1
|
||
}
|
||
|
||
// 遍历并去重路径
|
||
for _, folder := range ret {
|
||
if strings.HasPrefix(folder.Path, queryPath) {
|
||
// 提取下一级路径
|
||
path := folder.Path[len(queryParams.Path)+stepNum:]
|
||
pathArr := strings.Split(path, sch.Split)
|
||
if len(pathArr) > 0 {
|
||
// 拼接新的路径并去重
|
||
newPath := queryParams.Path + "/" + pathArr[0]
|
||
pathSet[newPath] = folder.CreateTime
|
||
}
|
||
}
|
||
}
|
||
|
||
// 将去重后的路径从 map 转换回切片
|
||
var result []uploadersdk.Folder
|
||
for path, createTime := range pathSet {
|
||
result = append(result, uploadersdk.Folder{
|
||
Path: path,
|
||
CreateTime: createTime,
|
||
})
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) QueryFolderByPackageID(ctx SQLContext, packageID cdssdk.PackageID) ([]uploadersdk.Folder, error) {
|
||
var ret []uploadersdk.Folder
|
||
if err := ctx.Table("folders").Where("package_id = ?", packageID).Find(&ret).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertFolder(ctx SQLContext, packageID cdssdk.PackageID, path string) error {
|
||
folder := uploadersdk.Folder{
|
||
PackageID: packageID,
|
||
Path: path + sch.Split,
|
||
CreateTime: time.Now(),
|
||
}
|
||
|
||
if err := ctx.Table("folders").Create(&folder).Error; err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) DeleteFolder(ctx SQLContext, packageID cdssdk.PackageID, path string) error {
|
||
path = sch.Split + path + sch.Split
|
||
err := ctx.Table("folders").Where("package_id = ? and path_name like ?", packageID, path+"%").Delete(&uploadersdk.Folder{}).Error
|
||
return err
|
||
}
|
||
|
||
func (db *UploadDataDB) QueryPackage(ctx SQLContext, queryParams sch.QueryData) ([]uploadersdk.Package, int, int, error) {
|
||
var ret []uploadersdk.PackageDAO
|
||
|
||
packageNamePattern := "%" + queryParams.PackageName + "%"
|
||
|
||
// 计算分页参数
|
||
limit := queryParams.PageSize
|
||
offset := (queryParams.CurrentPage - 1) * queryParams.PageSize
|
||
|
||
// 先查 total 数量
|
||
var totalNum int64
|
||
err := ctx.Table("package").
|
||
Where("user_id = ? AND data_type = ? AND package_name LIKE ?", queryParams.UserID, queryParams.DataType, packageNamePattern).
|
||
Count(&totalNum).Error
|
||
if err != nil {
|
||
return nil, 0, 0, err
|
||
}
|
||
|
||
// 再查分页数据
|
||
err = ctx.Table("package").
|
||
Where("user_id = ? AND data_type = ? AND package_name LIKE ? ORDER BY create_time DESC", queryParams.UserID, queryParams.DataType, packageNamePattern).
|
||
Preload("UploadedCluster").
|
||
Limit(limit).Offset(offset).
|
||
Find(&ret).Error
|
||
if err != nil {
|
||
return nil, 0, 0, err
|
||
}
|
||
|
||
// 转换 DAO -> Package
|
||
var res []uploadersdk.Package
|
||
for _, dao := range ret {
|
||
var uploadPriority sch.UploadPriority
|
||
if dao.UploadPriority != "" {
|
||
uploadPriority, err = serder.JSONToObjectEx[sch.UploadPriority]([]byte(dao.UploadPriority))
|
||
if err != nil {
|
||
return nil, 0, 0, err
|
||
}
|
||
}
|
||
|
||
var bindingInfo sch.DataBinding
|
||
if dao.Param != "" {
|
||
bindingInfo, err = serder.JSONToObjectEx[sch.DataBinding]([]byte(dao.Param))
|
||
if err != nil {
|
||
return nil, 0, 0, err
|
||
}
|
||
}
|
||
|
||
pkg := uploadersdk.Package{
|
||
UserID: dao.UserID,
|
||
PackageID: dao.PackageID,
|
||
PackageName: dao.PackageName,
|
||
BucketID: dao.BucketID,
|
||
DataType: dao.DataType,
|
||
BindingID: dao.BindingID,
|
||
UploadPriority: uploadPriority,
|
||
CreateTime: dao.CreateTime,
|
||
UploadedCluster: dao.UploadedCluster,
|
||
BindingInfo: bindingInfo,
|
||
}
|
||
res = append(res, pkg)
|
||
}
|
||
|
||
totalPages := int(totalNum) / queryParams.PageSize
|
||
if int(totalNum)%queryParams.PageSize != 0 {
|
||
totalPages++
|
||
}
|
||
|
||
return res, totalPages, int(totalNum), nil
|
||
}
|
||
|
||
func (db *UploadDataDB) QueryPackageByBindingID(ctx SQLContext, id uploadersdk.DataID) (*uploadersdk.Package, error) {
|
||
var dao uploadersdk.PackageDAO
|
||
err := ctx.Table("package").Where("binding_id = ?", id).Preload("Versions").Find(&dao).Error
|
||
|
||
if dao.PackageID == 0 {
|
||
var clonePkg uploadersdk.PackageCloneVO
|
||
|
||
// 获取子算法
|
||
err := ctx.Table("package_clone").
|
||
Where("binding_id = ?", id).
|
||
Find(&clonePkg).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if clonePkg.ClonePackageID == 0 {
|
||
return nil, fmt.Errorf("package not found")
|
||
}
|
||
|
||
var ret2 uploadersdk.PackageDAO
|
||
err = ctx.Table("package").Where("package_id = ?", clonePkg.ParentPackageID).Find(&ret2).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
ret2.PackageID = clonePkg.ClonePackageID
|
||
ret2.PackageName = clonePkg.Name
|
||
dao = ret2
|
||
}
|
||
|
||
// 将数据转换成 uploadersdk.Package
|
||
pkg := uploadersdk.Package{
|
||
UserID: dao.UserID,
|
||
PackageID: dao.PackageID,
|
||
PackageName: dao.PackageName,
|
||
BucketID: dao.BucketID,
|
||
DataType: dao.DataType,
|
||
//JsonData: dao.JsonData,
|
||
BindingID: dao.BindingID,
|
||
CreateTime: dao.CreateTime,
|
||
UploadedCluster: dao.UploadedCluster,
|
||
Versions: dao.Versions,
|
||
PackageType: dao.PackageType,
|
||
}
|
||
|
||
return &pkg, err
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertPackage(ctx SQLContext, newPackage uploadersdk.Package, clusters []uploadersdk.Cluster) error {
|
||
|
||
// 查询是否存在
|
||
if err := ctx.Table("package").Where("package_id = ?", newPackage.PackageID).First(&uploadersdk.PackageDAO{}).Error; err == nil {
|
||
return fmt.Errorf("package already exists")
|
||
}
|
||
|
||
// 将uploadPriority转成json string
|
||
uploadPriorityStr, err := json.Marshal(newPackage.UploadPriority)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to marshal uploadPriority: %w", err)
|
||
}
|
||
|
||
bindingStr, err := json.Marshal(newPackage.BindingInfo)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to marshal binding: %w", err)
|
||
}
|
||
|
||
bindingID := uploadersdk.DataID(-1)
|
||
if newPackage.BindingID != 0 {
|
||
bindingID = newPackage.BindingID
|
||
}
|
||
|
||
dao := uploadersdk.PackageDAO{
|
||
PackageID: newPackage.PackageID,
|
||
PackageName: newPackage.PackageName,
|
||
DataType: newPackage.DataType,
|
||
UserID: newPackage.UserID,
|
||
BucketID: newPackage.BucketID,
|
||
UploadPriority: string(uploadPriorityStr),
|
||
CreateTime: time.Now(),
|
||
BindingID: bindingID,
|
||
Param: string(bindingStr),
|
||
PackageType: newPackage.PackageType,
|
||
}
|
||
|
||
tx := ctx.Begin()
|
||
if err = tx.Table("package").Create(&dao).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
for _, cluster := range clusters {
|
||
if err = tx.Table("uploaded_cluster").Create(&cluster).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
tx.Commit()
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertPackageDAO(ctx SQLContext, newPackage uploadersdk.PackageDAO, clusters []uploadersdk.Cluster) error {
|
||
|
||
// 查询是否存在
|
||
if err := ctx.Table("package").Where("package_id = ?", newPackage.PackageID).First(&uploadersdk.PackageDAO{}).Error; err == nil {
|
||
return fmt.Errorf("package already exists")
|
||
}
|
||
|
||
tx := ctx.Begin()
|
||
if err := tx.Table("package").Create(&newPackage).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
for _, cluster := range clusters {
|
||
if err := tx.Table("uploaded_cluster").Create(&cluster).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
}
|
||
|
||
tx.Commit()
|
||
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) DeletePackage(ctx SQLContext, userID cdssdk.UserID, packageID cdssdk.PackageID) error {
|
||
// 开启事务
|
||
tx := ctx.Begin()
|
||
err := tx.Table("package").
|
||
Where("package_id = ? and user_id = ?", packageID, userID).
|
||
Delete(&uploadersdk.PackageDAO{}).Error
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
err = tx.Table("uploaded_cluster").
|
||
Where("package_id = ?", packageID).
|
||
Delete(&uploadersdk.Cluster{}).Error
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
err = tx.Table("folders").
|
||
Where("package_id = ?", packageID).
|
||
Delete(&uploadersdk.Folder{}).Error
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
err = tx.Table("package_clone").
|
||
Where("parent_package_id = ?", packageID).
|
||
Delete(&uploadersdk.PackageCloneDAO{}).Error
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
// 提交事务
|
||
tx.Commit()
|
||
return nil
|
||
}
|
||
|
||
// TODO 查询子算法语句有问题，待解决
|
||
func (db *UploadDataDB) QueryPackageByID(ctx SQLContext, ID cdssdk.PackageID) (uploadersdk.PackageDAO, error) {
|
||
var ret uploadersdk.PackageDAO
|
||
err := ctx.Table("package").Where("package_id = ?", ID).
|
||
Omit("Objects").Preload("Versions").
|
||
Preload("UploadedCluster").Find(&ret).Error
|
||
|
||
if err != nil {
|
||
return ret, err
|
||
}
|
||
|
||
// 判断是否查询到数据
|
||
if ret.PackageID == 0 {
|
||
sql := `
|
||
select
|
||
pv.clone_package_id as package_id,
|
||
pv.name as package_name,
|
||
pkg.user_id,
|
||
pkg.bucket_id,
|
||
pkg.data_type,
|
||
pkg.json_data,
|
||
pkg.binding_id,
|
||
pkg.upload_priority
|
||
from
|
||
(
|
||
select
|
||
*
|
||
from
|
||
package_clone
|
||
where
|
||
clone_package_id = ?
|
||
) as pv
|
||
left join(
|
||
select
|
||
*
|
||
from
|
||
package
|
||
) as pkg
|
||
on
|
||
pv.parent_package_id = pkg.package_id
|
||
`
|
||
err = ctx.Raw(sql, ID).Scan(&ret).Error
|
||
if err != nil {
|
||
return ret, err
|
||
}
|
||
if ret.PackageID == 0 {
|
||
return ret, fmt.Errorf("package not found")
|
||
}
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetAllStorage(ctx SQLContext) (map[schsdk.ClusterID]cdssdk.StorageID, error) {
|
||
var rows []uploadersdk.ClusterMapping
|
||
err := ctx.Table("cluster_mapping").Find(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 手动将查询结果转换为 map
|
||
ret := make(map[schsdk.ClusterID]cdssdk.StorageID)
|
||
for _, row := range rows {
|
||
ret[row.ClusterID] = row.StorageID
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetClusterMapping(ctx SQLContext) (map[schsdk.ClusterID]uploadersdk.ClusterMapping, error) {
|
||
var rows []uploadersdk.ClusterMapping
|
||
err := ctx.Table("cluster_mapping").Find(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 手动将查询结果转换为 map
|
||
ret := make(map[schsdk.ClusterID]uploadersdk.ClusterMapping)
|
||
for _, row := range rows {
|
||
ret[row.ClusterID] = row
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetClusterByIDs(ctx SQLContext, clusterIDs []schsdk.ClusterID) ([]uploadersdk.ClusterMapping, error) {
|
||
var ret []uploadersdk.ClusterMapping
|
||
|
||
if err := ctx.Table("cluster_mapping").Where("cluster_id in ?", clusterIDs).Find(&ret).Error; err != nil {
|
||
return ret, err
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetClusterByID(ctx SQLContext, clusterID schsdk.ClusterID) (*uploadersdk.ClusterMapping, error) {
|
||
var ret *uploadersdk.ClusterMapping
|
||
|
||
if err := ctx.Table("cluster_mapping").Where("cluster_id = ?", clusterID).Find(&ret).Error; err != nil {
|
||
return ret, err
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetClusterByName(ctx SQLContext, name string) (*uploadersdk.ClusterMapping, error) {
|
||
var ret *uploadersdk.ClusterMapping
|
||
|
||
if err := ctx.Table("cluster_mapping").Where("cluster_name = ?", name).Find(&ret).Error; err != nil {
|
||
return ret, err
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertUploadedCluster(ctx SQLContext, cluster uploadersdk.Cluster) error {
|
||
// 根据package_id和cluster_id做判断，如果已经存在，则不插入
|
||
if err := ctx.Table("uploaded_cluster").Where("package_id = ? and cluster_id = ?", cluster.PackageID, cluster.ClusterID).First(&uploadersdk.Cluster{}).Error; err == nil {
|
||
logger.Warnf("the cluster %v is already exists in uploaded_cluster", cluster.ClusterID)
|
||
return nil
|
||
}
|
||
if err := ctx.Table("uploaded_cluster").Create(&cluster).Error; err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) UpdatePackage(ctx SQLContext, packageID cdssdk.PackageID, pkg uploadersdk.PackageDAO) error {
|
||
|
||
// 更新ret中字段不为空的字段
|
||
if err := ctx.Table("package").Where("package_id = ?", packageID).Updates(&pkg).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
//if jsonData != "" {
|
||
// if err := ctx.Table("package").Where("package_id = ?", packageID).Update("json_data", jsonData).Error; err != nil {
|
||
// return err
|
||
// }
|
||
//}
|
||
//
|
||
//if bindingID != -1 {
|
||
// if err := ctx.Table("package").Where("package_id = ?", packageID).Update("binding_id", bindingID).Error; err != nil {
|
||
// return err
|
||
// }
|
||
//}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertClonePackage(ctx SQLContext, dao uploadersdk.PackageCloneDAO) error {
|
||
if err := ctx.Table("package_clone").Create(&dao).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetParentClonePackageByPkgID(ctx SQLContext, packageID cdssdk.PackageID) (*uploadersdk.PackageCloneVO, error) {
|
||
var res uploadersdk.PackageCloneVO
|
||
// 获取父算法
|
||
err := ctx.Table("package_clone").
|
||
Where("parent_package_id = ? and clone_package_id = ?", packageID, packageID).
|
||
Preload("ClusterMapping").
|
||
Find(&res).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
//var bindingData uploadersdk.Binding
|
||
//err = ctx.Table("bindings").Where("id = ?", res.ObjectID).Find(&bindingData).Error
|
||
//if err != nil {
|
||
// return nil, err
|
||
//}
|
||
//res.Name = bindingData.Name
|
||
|
||
return &res, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetChildrenClonePackageByPkgID(ctx SQLContext, userID cdssdk.UserID, packageID cdssdk.PackageID) ([]uploadersdk.PackageCloneVO, error) {
|
||
var clonePkgs []uploadersdk.PackageCloneVO
|
||
|
||
// 获取子算法
|
||
err := ctx.Table("package_clone").
|
||
Where("parent_package_id = ? and clone_package_id != ?", packageID, packageID).
|
||
Preload("ClusterMapping").
|
||
Find(&clonePkgs).Error
|
||
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 展示绑定名
|
||
var pkgs []uploadersdk.PackageCloneVO
|
||
for i := range clonePkgs {
|
||
if clonePkgs[i].BindingID == 0 {
|
||
continue
|
||
}
|
||
|
||
var bindingData uploadersdk.Binding
|
||
err = ctx.Table("bindings").Where("id = ? and user_id = ?", clonePkgs[i].BindingID, userID).Find(&bindingData).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
clonePkgs[i].Name = bindingData.Name
|
||
pkgs = append(pkgs, clonePkgs[i])
|
||
}
|
||
|
||
return pkgs, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetClonePackageByPkgID(ctx SQLContext, packageID cdssdk.PackageID) (*uploadersdk.PackageCloneVO, error) {
|
||
var res uploadersdk.PackageCloneVO
|
||
|
||
// 获取子算法
|
||
err := ctx.Table("package_clone").
|
||
Where("clone_package_id = ?", packageID).
|
||
Preload("ClusterMapping").
|
||
Find(&res).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return &res, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) RemoveClonePackage(ctx SQLContext, parentPackageIDs []cdssdk.PackageID, packageIDs []cdssdk.PackageID) error {
|
||
// 先查询
|
||
var clonePkgs []uploadersdk.PackageCloneDAO
|
||
if err := ctx.Table("package_clone").Where("parent_package_id in ? or clone_package_id in ?", parentPackageIDs, packageIDs).Find(&clonePkgs).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
if len(clonePkgs) == 0 {
|
||
return nil
|
||
}
|
||
|
||
var bindingIDs []int64
|
||
for _, pkg := range clonePkgs {
|
||
bindingIDs = append(bindingIDs, int64(pkg.BindingID))
|
||
}
|
||
|
||
err := db.DeleteBindingsByID(ctx, bindingIDs)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
err = ctx.Table("package_clone").Where("parent_package_id in ? or clone_package_id in ?", parentPackageIDs, packageIDs).Delete(&uploadersdk.PackageCloneDAO{}).Error
|
||
return err
|
||
}
|
||
|
||
func (db *UploadDataDB) GetCloneParentPackage(ctx SQLContext, userID cdssdk.UserID, dataType string) ([]uploadersdk.PackageCloneVO, error) {
|
||
var res []uploadersdk.PackageCloneVO
|
||
sql := `
|
||
select
|
||
cp.*
|
||
from
|
||
(
|
||
select
|
||
package_id
|
||
from
|
||
package
|
||
where
|
||
user_id = ?
|
||
and data_type = ?
|
||
) as pkg
|
||
left join (
|
||
select
|
||
*
|
||
from
|
||
package_clone
|
||
where
|
||
parent_package_id = clone_package_id
|
||
) as cp
|
||
on
|
||
pkg.package_id = cp.parent_package_id
|
||
where
|
||
cp.id is not null
|
||
`
|
||
|
||
err := ctx.Raw(sql, userID, dataType).Preload("cluster_mapping").Scan(&res).Error
|
||
|
||
// 展示绑定名
|
||
var pkgs []uploadersdk.PackageCloneVO
|
||
for i := range res {
|
||
var bindingData uploadersdk.Binding
|
||
err = ctx.Table("bindings").Where("id = ? and user_id = ?", res[i].BindingID, userID).Find(&bindingData).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 删除脏数据
|
||
if bindingData.Name == "" {
|
||
continue
|
||
}
|
||
|
||
res[i].Name = bindingData.Name
|
||
pkgs = append(pkgs, res[i])
|
||
}
|
||
|
||
return pkgs, err
|
||
}
|
||
|
||
func (db *UploadDataDB) GetClonePackageNum(ctx SQLContext, userID cdssdk.UserID, dataType string) (int64, int64, error) {
|
||
var parentNum, childNum int64
|
||
sql := `
|
||
select
|
||
count(distinct parent_package_id) as parent_num,
|
||
count(distinct clone_package_id) as child_num
|
||
from
|
||
package_clone
|
||
where
|
||
parent_package_id in (
|
||
select
|
||
package_id
|
||
from
|
||
package
|
||
where
|
||
user_id = ?
|
||
and data_type = ?
|
||
)
|
||
`
|
||
|
||
// 使用 Raw 查询单行数据
|
||
err := ctx.Raw(sql, userID, dataType).Row().Scan(&parentNum, &childNum)
|
||
if err != nil {
|
||
return 0, 0, err
|
||
}
|
||
|
||
return parentNum, childNum - parentNum, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertBlockchains(ctx SQLContext, blockchains []*uploadersdk.BlockChain) error {
|
||
|
||
if err := ctx.Table("block_chain").Create(&blockchains).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetPrivateBindings(ctx SQLContext, userID cdssdk.UserID, dataType string, filters sch.QueryBindingFilters) ([]uploadersdk.Binding, error) {
|
||
var rows []uploadersdk.Binding
|
||
|
||
// 构建基础查询条件
|
||
query := ctx.Table("bindings").Where("user_id = ? and data_type = ?", userID, dataType)
|
||
|
||
// 根据 filters 动态添加查询条件
|
||
if filters.Status != "" {
|
||
query = query.Where("status = ?", filters.Status)
|
||
}
|
||
if filters.Name != "" {
|
||
query = query.Where("name like ?", "%"+filters.Name+"%")
|
||
}
|
||
|
||
// 执行查询
|
||
err := query.Find(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return rows, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetApplyBindings(ctx SQLContext, userID cdssdk.UserID, level string, dataType string, filters sch.QueryBindingFilters) ([]uploadersdk.BindingAccessData, error) {
|
||
var ret []uploadersdk.BindingAccessData
|
||
|
||
// 基础 SQL 构建（外层不含 WHERE 子句）
|
||
sql := `
|
||
select
|
||
bd.id as ID,
|
||
bd.user_id,
|
||
bd.name,
|
||
bd.data_type,
|
||
bd.content,
|
||
bd.created_at,
|
||
ar.applicant_id,
|
||
ar.status,
|
||
usr.username,
|
||
usr.sso_id
|
||
from
|
||
(
|
||
select * from bindings
|
||
where access_level = ? and data_type = ? and user_id != ?
|
||
) as bd
|
||
left join (
|
||
select data_owner_id, binding_id, applicant_id, status
|
||
from access_requests
|
||
where applicant_id = ? and status != ?
|
||
) as ar on bd.user_id = ar.data_owner_id and bd.id = ar.binding_id
|
||
left join users as usr on bd.user_id = usr.id
|
||
`
|
||
|
||
// 初始参数列表
|
||
args := []interface{}{level, dataType, userID, userID, sch.ExpiredStatus}
|
||
|
||
// 动态 WHERE 条件
|
||
var whereClauses []string
|
||
if filters.Status != "" {
|
||
whereClauses = append(whereClauses, "ar.status = ?")
|
||
args = append(args, filters.Status)
|
||
}
|
||
if filters.Name != "" {
|
||
whereClauses = append(whereClauses, "bd.name LIKE ?")
|
||
args = append(args, "%"+filters.Name+"%")
|
||
}
|
||
|
||
// 拼接 WHERE 子句
|
||
if len(whereClauses) > 0 {
|
||
sql += " WHERE " + strings.Join(whereClauses, " AND ")
|
||
}
|
||
|
||
// 执行查询
|
||
err := ctx.Raw(sql, args...).Scan(&ret).Debug().Error
|
||
if err != nil {
|
||
return ret, err
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetPublicBindings(ctx SQLContext, level string, dataType string, userID cdssdk.UserID, filters sch.QueryBindingFilters) ([]uploadersdk.BindingAccessData, error) {
|
||
var ret []uploadersdk.BindingAccessData
|
||
|
||
sql := `
|
||
select
|
||
bd.id as ID,
|
||
bd.user_id,
|
||
bd.name,
|
||
bd.data_type,
|
||
bd.content,
|
||
bd.access_level,
|
||
bd.created_at,
|
||
ar.applicant_id,
|
||
ar.status,
|
||
usr.username,
|
||
usr.sso_id
|
||
from
|
||
(
|
||
select * from bindings
|
||
where access_level = ? and data_type = ? and user_id != ?
|
||
) as bd
|
||
left join (
|
||
select data_owner_id as user_id, applicant_id, status, binding_id
|
||
from access_requests
|
||
) as ar on bd.user_id = ar.user_id
|
||
and bd.id = ar.binding_id
|
||
left join users as usr on bd.user_id = usr.id
|
||
`
|
||
|
||
// 初始参数
|
||
args := []interface{}{level, dataType, userID}
|
||
|
||
// 构建 WHERE 子句
|
||
var whereClauses []string
|
||
if filters.Status != "" {
|
||
whereClauses = append(whereClauses, "ar.status = ?")
|
||
args = append(args, filters.Status)
|
||
}
|
||
if filters.Name != "" {
|
||
whereClauses = append(whereClauses, "bd.name LIKE ?")
|
||
args = append(args, "%"+filters.Name+"%")
|
||
}
|
||
|
||
// 拼接 WHERE 子句
|
||
if len(whereClauses) > 0 {
|
||
sql += " WHERE " + strings.Join(whereClauses, " AND ")
|
||
}
|
||
|
||
// 执行查询
|
||
err := ctx.Raw(sql, args...).Scan(&ret).Debug().Error
|
||
if err != nil {
|
||
return ret, err
|
||
}
|
||
return ret, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetBindingByID(ctx SQLContext, userID cdssdk.UserID, ID uploadersdk.DataID) (*uploadersdk.Binding, error) {
|
||
|
||
var rows uploadersdk.Binding
|
||
err := ctx.Table("bindings").Where("id = ? and user_id = ?", ID, userID).Find(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &rows, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetBindingByIDs(ctx SQLContext, userID cdssdk.UserID, IDs []int64) ([]uploadersdk.BindingDAO, error) {
|
||
|
||
var rows []uploadersdk.BindingDAO
|
||
err := ctx.Table("bindings").Where("id in ? and user_id = ?", IDs, userID).
|
||
Preload("BindingCluster").Find(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) DeleteBindingsByID(ctx SQLContext, IDs []int64) error {
|
||
tx := ctx.Begin()
|
||
if err := tx.Table("package").Where("binding_id in ?", IDs).Update("binding_id", -1).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
err := tx.Table("access_requests").Where("binding_id in ? and status = ?", IDs, sch.ApprovedStatus).Update("status", sch.ExpiredStatus).Error
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
if err := tx.Table("bindings").Where("id in ?", IDs).Delete(&uploadersdk.Binding{}).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
if err := tx.Table("binding_cluster").Where("binding_id in ?", IDs).Delete(&uploadersdk.BindingCluster{}).Error; err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
|
||
tx.Commit()
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetBindingByName(ctx SQLContext, userID cdssdk.UserID, name string, dataType string) (*uploadersdk.Binding, error) {
|
||
// 根据name查询是否存在
|
||
var existData uploadersdk.Binding
|
||
err := ctx.Table("bindings").Where("user_id = ? and name = ? and data_type = ?", userID, name, dataType).Find(&existData).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &existData, nil
|
||
}
|
||
|
||
// GetBindingByDataType 根据dataType查询是否存在
|
||
func (db *UploadDataDB) GetBindingByDataType(ctx SQLContext, userID cdssdk.UserID, dataType string) ([]uploadersdk.Binding, error) {
|
||
var existData []uploadersdk.Binding
|
||
// 如果没有用户ID，则查所有用户的统计信息
|
||
if userID == 0 {
|
||
err := ctx.Table("bindings").Where("data_type = ?", dataType).Find(&existData).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return existData, nil
|
||
}
|
||
err := ctx.Table("bindings").Where("user_id = ? and data_type = ?", userID, dataType).Find(&existData).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return existData, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetBindingCluster(ctx SQLContext, bindingID uploadersdk.DataID) ([]uploadersdk.BindingCluster, error) {
|
||
// 根据name查询是否存在
|
||
var clusters []uploadersdk.BindingCluster
|
||
err := ctx.Table("binding_cluster").Where("binding_id = ?", bindingID).Find(&clusters).Error
|
||
|
||
return clusters, err
|
||
}
|
||
|
||
func (db *UploadDataDB) UpdateBindingCluster(ctx SQLContext, bindingID uploadersdk.DataID, clusterID schsdk.ClusterID, jsonData string) error {
|
||
return ctx.Table("binding_cluster").Where("binding_id = ? and cluster_id = ?", bindingID, clusterID).Update("json_data", jsonData).Error
|
||
}
|
||
|
||
func (db *UploadDataDB) InsertOrUpdateBinding(ctx SQLContext, data uploadersdk.Binding, bindingClusters []uploadersdk.BindingCluster, packageID cdssdk.PackageID) (*schsdk.DataID, error) {
|
||
|
||
tx := ctx.Begin()
|
||
|
||
// 插入BindingData
|
||
if data.ID == -1 {
|
||
data.ID = 0
|
||
err := tx.Table("bindings").Create(&data).Error
|
||
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
if data.ID == 0 {
|
||
tx.Rollback()
|
||
return nil, fmt.Errorf("insert failed: ID is 0")
|
||
}
|
||
|
||
// 插入bindingCluster
|
||
for _, bc := range bindingClusters {
|
||
bc.BindingID = data.ID
|
||
err := tx.Table("binding_cluster").Create(&bc).Error
|
||
if err != nil {
|
||
tx.Rollback()
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
// 更新package的binding_id
|
||
pkgDao := uploadersdk.PackageDAO{
|
||
BindingID: data.ID,
|
||
}
|
||
if err := tx.Table("package").Where("package_id = ?", packageID).Updates(&pkgDao).Error; err != nil {
|
||
tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
// 更新packageClone的binding_id（如果有）
|
||
pkgCloneDao := uploadersdk.PackageCloneDAO{
|
||
BindingID: data.ID,
|
||
}
|
||
if err := tx.Table("package_clone").Where("clone_package_id = ?", packageID).Updates(&pkgCloneDao).Error; err != nil {
|
||
tx.Rollback()
|
||
return nil, err
|
||
}
|
||
|
||
tx.Commit()
|
||
|
||
return (*schsdk.DataID)(&data.ID), nil
|
||
}
|
||
|
||
func (db *UploadDataDB) UpdateBindingAccess(ctx SQLContext, id uploadersdk.DataID, level string) error {
|
||
if err := ctx.Table("bindings").Where("id = ?", id).Update("access_level", level).Error; err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetImageByClusterID(ctx SQLContext, clusterIDs []string, cardTypes []string, imageIDs []schsdk.ImageID) ([]sch.Image, error) {
|
||
var images []sch.Image
|
||
|
||
query := ctx.Table("Image").
|
||
Preload("ClusterImage").
|
||
Preload("ClusterImage.ClusterImageCard")
|
||
|
||
// 处理 imageIDs 过滤
|
||
if len(imageIDs) > 0 {
|
||
query = query.Where("ImageID IN ?", imageIDs)
|
||
}
|
||
|
||
// 处理 clusterIDs 过滤
|
||
if len(clusterIDs) > 0 {
|
||
query = query.Where("ImageID IN (SELECT image_id FROM cluster_image WHERE cluster_id IN ?)", clusterIDs)
|
||
}
|
||
|
||
// 处理 cardTypes 过滤
|
||
if len(cardTypes) > 0 {
|
||
query = query.Where("ImageID IN (SELECT image_id FROM cluster_image WHERE origin_image_id IN (SELECT origin_image_id FROM cluster_image_card WHERE card IN ?))", cardTypes)
|
||
}
|
||
|
||
err := query.Find(&images).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return images, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetImageByID(ctx SQLContext, imageID schsdk.ImageID) (*sch.Image, error) {
|
||
var images sch.Image
|
||
err := ctx.Table("Image").
|
||
Preload("ClusterImage").
|
||
Preload("ClusterImage.ClusterImageCard").
|
||
Where("ImageID = ?", imageID).
|
||
Find(&images).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return &images, nil
|
||
}
|
||
|
||
func (db *UploadDataDB) GetImageNum(ctx SQLContext) (int64, error) {
|
||
var num int64
|
||
err := ctx.Table("Image").Count(&num).Error
|
||
return num, err
|
||
}
|
||
|
||
func (db *UploadDataDB) GetPublicImageByID(ctx SQLContext, IDs []string) ([]sch.Image, error) {
|
||
var images []sch.Image
|
||
err := ctx.Table("Image").
|
||
Preload("ClusterImage").
|
||
Preload("ClusterImage.ClusterImageCard").
|
||
Where("ImageID IN ?", IDs).
|
||
Find(&images).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return images, nil
|
||
}
|