mongo开发工具类增加增删改操作日志写入redis消息逻辑

mongo/mongo.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"gitee.com/red-future---jilin-g/common/log/model/dto"
 	"github.com/gogf/gf/v2/container/gvar"
 
 	"gitee.com/red-future---jilin-g/common/consts"
@@ -403,12 +404,25 @@ func (m *MongoDB) Delete(ctx context.Context, filter bson.M, collection string,
 		return
 	}
 	filter["tenantId"] = user.TenantId
+	var rows []interface{}
+	if err = m.Find(ctx, filter, &rows, collection); err != nil {
+		return
+	}
 	r, err := db.Collection(collection).DeleteMany(ctx, filter, opts...)
 	if err != nil {
 		return
 	}
 	count = r.DeletedCount
 	err = cleanRedis(ctx, filter, user.TenantId, collection)
+	serverName := g.Cfg().MustGet(ctx, "server.name").String()
+	logRedisKey := fmt.Sprintf("log:%s", serverName)
+	if _, err = redis.AddToStream(ctx, logRedisKey, &dto.RecordCreateLogReq{
+		ServiceName: serverName,
+		Collection:  collection,
+		Data:        rows,
+	}); err != nil {
+		glog.Error(ctx, "mongoLog-AddToStream err: %v", err)
+	}
 	return
 }
 
@@ -432,11 +446,24 @@ func (m *MongoDB) Update(ctx context.Context, filter bson.M, update bson.M, coll
 	}
 	setDoc["updatedAt"] = gtime.Now().Time
 	update = bson.M{"$set": setDoc}
+	var rows []interface{}
+	if err = m.Find(ctx, filter, &rows, collection); err != nil {
+		return
+	}
 	result, err = db.Collection(collection).UpdateMany(ctx, filter, update, opts...)
 	if err != nil {
 		return
 	}
 	err = cleanRedis(ctx, filter, user.TenantId, collection)
+	serverName := g.Cfg().MustGet(ctx, "server.name").String()
+	logRedisKey := fmt.Sprintf("log:%s", serverName)
+	if _, err = redis.AddToStream(ctx, logRedisKey, &dto.RecordCreateLogReq{
+		ServiceName: serverName,
+		Collection:  collection,
+		Data:        rows,
+	}); err != nil {
+		glog.Error(ctx, "mongoLog-AddToStream err: %v", err)
+	}
 	return
 }
 
@@ -592,6 +619,30 @@ func (m *MongoDB) Insert(ctx context.Context, documents []interface{}, collectio
 	}
 	ids = r.InsertedIDs
 	err = cleanRedis(ctx, bson.M{}, user.TenantId, collection)
+	//写日志
+	serverName := g.Cfg().MustGet(ctx, "server.name").String()
+	logRedisKey := fmt.Sprintf("log:%s", serverName)
+	if len(ids) == 0 {
+		return
+	}
+	rows := make([]interface{}, 0, len(ids))
+	if len(ids) == 1 {
+		doc := gconv.Map(documents[0])
+		doc["id"] = ids[0]
+		rows = append(rows, doc)
+	} else {
+		filter := bson.M{"_id": bson.M{"$in": ids}}
+		if err = m.Find(ctx, filter, &rows, collection); err != nil {
+			return
+		}
+	}
+	if _, err = redis.AddToStream(ctx, logRedisKey, &dto.RecordCreateLogReq{
+		ServiceName: serverName,
+		Collection:  collection,
+		Data:        rows,
+	}); err != nil {
+		glog.Error(ctx, "mongoLog-AddToStream err: %v", err)
+	}
 	return
 }