96 lines
3.0 KiB
Go
96 lines
3.0 KiB
Go
package dao
|
||
|
||
import (
|
||
"context"
|
||
commonMongo "gitea.com/red-future/common/db/mongo"
|
||
"github.com/gogf/gf/v2/errors/gerror"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/glog"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
"go.mongodb.org/mongo-driver/v2/bson"
|
||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||
)
|
||
|
||
// MongoDAO MongoDB原生查询（不需要token验证）
|
||
var MongoDAO = &mongoDAO{}
|
||
|
||
type mongoDAO struct{}
|
||
|
||
// SaveOrUpdate 原生批量增加或修改
|
||
func (d *mongoDAO) SaveOrUpdate(ctx context.Context, filter []bson.M, update []bson.M, collection string, opts ...options.Lister[options.UpdateManyOptions]) (result *mongo.BulkWriteResult, err error) {
|
||
db := commonMongo.GetDB()
|
||
if len(filter) == 0 || len(update) == 0 {
|
||
err = gerror.New("缺少查询条件或更新数据")
|
||
return
|
||
}
|
||
if len(filter) != len(update) {
|
||
err = gerror.New("查询条件和更新数据的数量必须一致")
|
||
return
|
||
}
|
||
// 构建批量操作模型
|
||
var models []mongo.WriteModel
|
||
for i := 0; i < len(filter); i++ {
|
||
if g.IsEmpty(filter[i]["tenantId"]) && g.IsEmpty(gconv.Map(update[i]["$set"])["updater"]) {
|
||
return nil, gerror.New("tenantId不能为空")
|
||
}
|
||
if g.IsEmpty(filter[i]["updater"]) && g.IsEmpty(gconv.Map(update[i]["$set"])["updater"]) {
|
||
return nil, gerror.New("updater不能为空")
|
||
}
|
||
// 处理过滤器
|
||
filter[i]["isDeleted"] = false
|
||
// 处理更新数据
|
||
if setDoc, exists := update[i]["$set"].(bson.M); exists {
|
||
setDoc["updater"] = gconv.Map(update[i]["$set"])["updater"]
|
||
setDoc["updatedAt"] = gtime.Now().Time
|
||
} else {
|
||
// 如果没有$set字段，则创建一个
|
||
setDoc := bson.M{}
|
||
setDoc["updater"] = gconv.Map(update[i]["$set"])["updater"]
|
||
setDoc["updatedAt"] = gtime.Now().Time
|
||
update[i]["$set"] = setDoc
|
||
}
|
||
// 创建更新操作模型
|
||
updateModel := mongo.NewUpdateOneModel()
|
||
updateModel.SetFilter(filter[i])
|
||
updateModel.SetUpdate(update[i])
|
||
updateModel.SetUpsert(true) // 默认不插入新文档
|
||
// 处理选项参数
|
||
if len(opts) > 0 {
|
||
for _, opt := range opts {
|
||
var updateOpts options.UpdateManyOptions
|
||
optFuncs := opt.List()
|
||
for _, fn := range optFuncs {
|
||
fn(&updateOpts)
|
||
}
|
||
if updateOpts.Upsert != nil {
|
||
updateModel.SetUpsert(*updateOpts.Upsert)
|
||
}
|
||
}
|
||
}
|
||
models = append(models, updateModel)
|
||
}
|
||
// 执行批量操作，无序执行提高性能
|
||
bulkOpts := options.BulkWrite().SetOrdered(false)
|
||
bulkResult, err := db.Collection(collection).BulkWrite(ctx, models, bulkOpts)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 清理相关缓存
|
||
for i := 0; i < len(filter); i++ {
|
||
var tenantId any
|
||
if g.IsEmpty(filter[i]["tenantId"]) {
|
||
tenantId = filter[i]["tenantId"]
|
||
}
|
||
if g.IsEmpty(gconv.Map(update[i]["$set"])["tenantId"]) {
|
||
tenantId = gconv.Map(update[i]["$set"])["tenantId"]
|
||
}
|
||
err = commonMongo.DB().CleanRedis(ctx, filter[i], tenantId, collection)
|
||
if err != nil {
|
||
glog.Warning(ctx, "清理Redis缓存失败:", err)
|
||
}
|
||
}
|
||
return bulkResult, nil
|
||
}
|