IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-04 09:05:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

BatchInsertBlock

Browse Source
This commit is contained in:
withchao
2023-05-25 14:57:15 +08:00
parent b8c3400145
commit 30d8276986
4 changed files with 344 additions and 774 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/db/controller/msg.go
+168 -158
View File
@@ -26,8 +26,6 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"github.com/go-redis/redis/v8"
"go.mongodb.org/mongo-driver/mongo"
"google.golang.org/protobuf/proto"
)
type CommonMsgDatabase interface {
@@ -148,6 +146,7 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
return nil
}
num := db.msg.GetSingleGocMsgNum()
num = 100
if msgList[0].Msg != nil {
firstSeq = msgList[0].Msg.Seq
}
@@ -164,13 +163,13 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
err error
)
if msg.Msg != nil {
res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "msg", msgList[0].Msg)
res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "msg", msg.Msg)
} else if msg.Revoke != nil {
res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "revoke", msgList[0].Revoke)
res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "revoke", msg.Revoke)
} else if msg.DelList != nil {
res, err = db.msgDocDatabase.PushUnique(ctx, docID, index, "del_list", msgList[0].DelList)
res, err = db.msgDocDatabase.PushUnique(ctx, docID, index, "del_list", msg.DelList)
} else if msg.ReadList != nil {
res, err = db.msgDocDatabase.PushUnique(ctx, docID, index, "read_list", msgList[0].ReadList)
res, err = db.msgDocDatabase.PushUnique(ctx, docID, index, "read_list", msg.ReadList)
} else {
return false, errs.ErrArgs.Wrap("msg all field is nil")
}
@@ -199,12 +198,21 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
}
var insert int
for j := i; j < len(msgList); j++ {
seq = firstSeq + int64(j)
if getDocID(seq) != docID {
break
}
insert++
doc.Msg[getIndex(seq)] = *msgList[j]
}
for i, model := range doc.Msg {
if model.DelList == nil {
doc.Msg[i].DelList = []string{}
}
if model.ReadList == nil {
doc.Msg[i].ReadList = []string{}
}
}
if err := db.msgDocDatabase.Create(ctx, &doc); err != nil {
if mongo.IsDuplicateKeyError(err) {
i--
@@ -220,91 +228,91 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
}
func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
num := db.msg.GetSingleGocMsgNum()
currentIndex := currentMaxSeq / num
var blockMsgs []*[]*sdkws.MsgData
for i, data := range msgList {
data.Seq = currentMaxSeq + int64(i+1)
index := data.Seq/num - currentIndex
if i == 0 && index == 1 {
index--
currentIndex++
}
var block *[]*sdkws.MsgData
if len(blockMsgs) == int(index) {
var size int64
if i == 0 {
size = num - data.Seq%num
} else {
temp := int64(len(msgList)-len(*blockMsgs[0])) - int64(len(blockMsgs)-1)*num
if temp >= num {
size = num
} else {
size = temp % num
}
}
temp := make([]*sdkws.MsgData, 0, size)
block = &temp
blockMsgs = append(blockMsgs, block)
} else {
block = blockMsgs[index]
}
*block = append(*block, msgList[i])
}
create := currentMaxSeq == 0 || ((*blockMsgs[0])[0].Seq%num == 0)
if !create {
exist, err := db.msgDocDatabase.IsExistDocID(ctx, db.msg.IndexDocID(conversationID, currentIndex))
if err != nil {
return err
}
create = !exist
}
for i, msgs := range blockMsgs {
docID := db.msg.IndexDocID(conversationID, currentIndex+int64(i))
if create || i != 0 { // 插入
doc := unRelationTb.MsgDocModel{
DocID: docID,
Msg: make([]unRelationTb.MsgInfoModel, num),
}
for i := 0; i < len(doc.Msg); i++ {
doc.Msg[i].ReadList = []string{}
doc.Msg[i].DelList = []string{}
}
for _, msg := range *msgs {
data, err := proto.Marshal(msg)
if err != nil {
return err
}
doc.Msg[msg.Seq%num] = unRelationTb.MsgInfoModel{
SendTime: msg.SendTime,
Msg: data,
ReadList: []string{},
DelList: []string{},
}
}
if err := db.msgDocDatabase.Create(ctx, &doc); err != nil {
prome.Inc(prome.MsgInsertMongoFailedCounter)
return utils.Wrap(err, "")
}
prome.Inc(prome.MsgInsertMongoSuccessCounter)
} else { // 修改
for _, msg := range *msgs {
data, err := proto.Marshal(msg)
if err != nil {
return err
}
info := unRelationTb.MsgInfoModel{
SendTime: msg.SendTime,
Msg: data,
}
if err := db.msgDocDatabase.UpdateMsg(ctx, docID, msg.Seq%num, &info); err != nil {
prome.Inc(prome.MsgInsertMongoFailedCounter)
return err
}
prome.Inc(prome.MsgInsertMongoSuccessCounter)
}
}
}
//num := db.msg.GetSingleGocMsgNum()
//currentIndex := currentMaxSeq / num
//var blockMsgs []*[]*sdkws.MsgData
//for i, data := range msgList {
// data.Seq = currentMaxSeq + int64(i+1)
// index := data.Seq/num - currentIndex
// if i == 0 && index == 1 {
// index--
// currentIndex++
// }
// var block *[]*sdkws.MsgData
// if len(blockMsgs) == int(index) {
// var size int64
// if i == 0 {
// size = num - data.Seq%num
// } else {
// temp := int64(len(msgList)-len(*blockMsgs[0])) - int64(len(blockMsgs)-1)*num
// if temp >= num {
// size = num
// } else {
// size = temp % num
// }
// }
// temp := make([]*sdkws.MsgData, 0, size)
// block = &temp
// blockMsgs = append(blockMsgs, block)
// } else {
// block = blockMsgs[index]
// }
// *block = append(*block, msgList[i])
//}
//create := currentMaxSeq == 0 || ((*blockMsgs[0])[0].Seq%num == 0)
//if !create {
// exist, err := db.msgDocDatabase.IsExistDocID(ctx, db.msg.IndexDocID(conversationID, currentIndex))
// if err != nil {
// return err
// }
// create = !exist
//}
//for i, msgs := range blockMsgs {
// docID := db.msg.IndexDocID(conversationID, currentIndex+int64(i))
// if create || i != 0 { // 插入
// doc := unRelationTb.MsgDocModel{
// DocID: docID,
// Msg: make([]unRelationTb.MsgInfoModel, num),
// }
// for i := 0; i < len(doc.Msg); i++ {
// doc.Msg[i].ReadList = []string{}
// doc.Msg[i].DelList = []string{}
// }
// for _, msg := range *msgs {
// data, err := proto.Marshal(msg)
// if err != nil {
// return err
// }
// doc.Msg[msg.Seq%num] = unRelationTb.MsgInfoModel{
// SendTime: msg.SendTime,
// Msg: data,
// ReadList: []string{},
// DelList: []string{},
// }
// }
// if err := db.msgDocDatabase.Create(ctx, &doc); err != nil {
// prome.Inc(prome.MsgInsertMongoFailedCounter)
// return utils.Wrap(err, "")
// }
// prome.Inc(prome.MsgInsertMongoSuccessCounter)
// } else { // 修改
// for _, msg := range *msgs {
// data, err := proto.Marshal(msg)
// if err != nil {
// return err
// }
// info := unRelationTb.MsgInfoModel{
// SendTime: msg.SendTime,
// Msg: data,
// }
// if err := db.msgDocDatabase.UpdateMsg(ctx, docID, msg.Seq%num, &info); err != nil {
// prome.Inc(prome.MsgInsertMongoFailedCounter)
// return err
// }
// prome.Inc(prome.MsgInsertMongoSuccessCounter)
// }
// }
//}
return nil
}
@@ -408,10 +416,11 @@ func (db *commonMsgDatabase) GetOldestMsg(ctx context.Context, conversationID st
func (db *commonMsgDatabase) unmarshalMsg(msgInfo *unRelationTb.MsgInfoModel) (msgPb *sdkws.MsgData, err error) {
msgPb = &sdkws.MsgData{}
err = proto.Unmarshal(msgInfo.Msg, msgPb)
if err != nil {
return nil, utils.Wrap(err, "")
}
// todo: unmarshal
//err = proto.Unmarshal(msgInfo.Msg, msgPb)
//if err != nil {
// return nil, utils.Wrap(err, "")
//}
return msgPb, nil
}
@@ -644,69 +653,70 @@ func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
// recursion 删除list并且返回设置的最小seq
func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversationID string, index int64, delStruct *delMsgRecursionStruct, remainTime int64) (int64, error) {
// find from oldest list
msgs, err := db.msgDocDatabase.GetMsgsByIndex(ctx, conversationID, index)
if err != nil || msgs.DocID == "" {
if err != nil {
if err == unrelation.ErrMsgListNotExist {
log.ZDebug(ctx, "deleteMsgRecursion ErrMsgListNotExist", "conversationID", conversationID, "index:", index)
} else {
log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
}
}
// 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList), 结束递归
err = db.msgDocDatabase.Delete(ctx, delStruct.delDocIDs)
if err != nil {
return 0, err
}
return delStruct.getSetMinSeq() + 1, nil
}
log.ZDebug(ctx, "doc info", "conversationID", conversationID, "index", index, "docID", msgs.DocID, "len", len(msgs.Msg))
if int64(len(msgs.Msg)) > db.msg.GetSingleGocMsgNum() {
log.ZWarn(ctx, "msgs too large", nil, "lenth", len(msgs.Msg), "docID:", msgs.DocID)
}
if msgs.Msg[len(msgs.Msg)-1].SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() && msgs.IsFull() {
delStruct.delDocIDs = append(delStruct.delDocIDs, msgs.DocID)
lastMsgPb := &sdkws.MsgData{}
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
if err != nil {
log.ZError(ctx, "proto.Unmarshal failed", err, "index", len(msgs.Msg)-1, "docID", msgs.DocID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
delStruct.minSeq = lastMsgPb.Seq
} else {
var hasMarkDelFlag bool
for i, msg := range msgs.Msg {
if msg.SendTime != 0 {
msgPb := &sdkws.MsgData{}
err = proto.Unmarshal(msg.Msg, msgPb)
if err != nil {
log.ZError(ctx, "proto.Unmarshal failed", err, "index", i, "docID", msgs.DocID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
if utils.GetCurrentTimestampByMill() > msg.SendTime+(remainTime*1000) {
msgPb.Status = constant.MsgDeleted
bytes, _ := proto.Marshal(msgPb)
msg.Msg = bytes
msg.SendTime = 0
hasMarkDelFlag = true
} else {
// 到本条消息不需要删除, minSeq置为这条消息的seq
if err := db.msgDocDatabase.Delete(ctx, delStruct.delDocIDs); err != nil {
return 0, err
}
if hasMarkDelFlag {
if err := db.msgDocDatabase.UpdateOneDoc(ctx, msgs); err != nil {
return delStruct.getSetMinSeq(), err
}
}
return msgPb.Seq, nil
}
}
}
}
// 继续递归 index+1
seq, err := db.deleteMsgRecursion(ctx, conversationID, index+1, delStruct, remainTime)
return seq, err
//msgs, err := db.msgDocDatabase.GetMsgsByIndex(ctx, conversationID, index)
//if err != nil || msgs.DocID == "" {
// if err != nil {
// if err == unrelation.ErrMsgListNotExist {
// log.ZDebug(ctx, "deleteMsgRecursion ErrMsgListNotExist", "conversationID", conversationID, "index:", index)
// } else {
// log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
// }
// }
// // 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList), 结束递归
// err = db.msgDocDatabase.Delete(ctx, delStruct.delDocIDs)
// if err != nil {
// return 0, err
// }
// return delStruct.getSetMinSeq() + 1, nil
//}
//log.ZDebug(ctx, "doc info", "conversationID", conversationID, "index", index, "docID", msgs.DocID, "len", len(msgs.Msg))
//if int64(len(msgs.Msg)) > db.msg.GetSingleGocMsgNum() {
// log.ZWarn(ctx, "msgs too large", nil, "lenth", len(msgs.Msg), "docID:", msgs.DocID)
//}
//if msgs.Msg[len(msgs.Msg)-1].SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() && msgs.IsFull() {
// delStruct.delDocIDs = append(delStruct.delDocIDs, msgs.DocID)
// lastMsgPb := &sdkws.MsgData{}
// err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
// if err != nil {
// log.ZError(ctx, "proto.Unmarshal failed", err, "index", len(msgs.Msg)-1, "docID", msgs.DocID)
// return 0, utils.Wrap(err, "proto.Unmarshal failed")
// }
// delStruct.minSeq = lastMsgPb.Seq
//} else {
// var hasMarkDelFlag bool
// for i, msg := range msgs.Msg {
// if msg.SendTime != 0 {
// msgPb := &sdkws.MsgData{}
// err = proto.Unmarshal(msg.Msg, msgPb)
// if err != nil {
// log.ZError(ctx, "proto.Unmarshal failed", err, "index", i, "docID", msgs.DocID)
// return 0, utils.Wrap(err, "proto.Unmarshal failed")
// }
// if utils.GetCurrentTimestampByMill() > msg.SendTime+(remainTime*1000) {
// msgPb.Status = constant.MsgDeleted
// bytes, _ := proto.Marshal(msgPb)
// msg.Msg = bytes
// msg.SendTime = 0
// hasMarkDelFlag = true
// } else {
// // 到本条消息不需要删除, minSeq置为这条消息的seq
// if err := db.msgDocDatabase.Delete(ctx, delStruct.delDocIDs); err != nil {
// return 0, err
// }
// if hasMarkDelFlag {
// if err := db.msgDocDatabase.UpdateOneDoc(ctx, msgs); err != nil {
// return delStruct.getSetMinSeq(), err
// }
// }
// return msgPb.Seq, nil
// }
// }
// }
//}
//// 继续递归 index+1
//seq, err := db.deleteMsgRecursion(ctx, conversationID, index+1, delStruct, remainTime)
//return seq, err
return 0, nil
}
func (db *commonMsgDatabase) CleanUpUserConversationsMsgs(ctx context.Context, user string, conversationIDs []string) {
pkg/common/db/controller/msg_test.go
+68 -517
View File
@@ -2,13 +2,12 @@ package controller
import (
"context"
"encoding/json"
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
"go.mongodb.org/mongo-driver/bson"
"math/rand"
"sort"
"strconv"
"sync"
"testing"
@@ -126,523 +125,75 @@ func Test_BatchInsertChat2DB(t *testing.T) {
}
func TestName(t *testing.T) {
s := ` [
189,
498,
310,
163,
313,
335,
327,
342,
123,
97,
4,
362,
210,
298,
436,
9,
369,
432,
132,
69,
248,
93,
91,
112,
145,
194,
84,
443,
179,
241,
257,
237,
169,
460,
33,
441,
126,
187,
390,
402,
51,
35,
455,
175,
389,
61,
309,
467,
492,
453,
159,
276,
165,
417,
173,
157,
12,
209,
269,
36,
226,
356,
92,
267,
482,
318,
219,
119,
176,
245,
74,
13,
450,
196,
215,
28,
167,
366,
442,
201,
341,
68,
2,
484,
328,
44,
423,
403,
105,
109,
480,
271,
134,
336,
299,
148,
365,
135,
277,
87,
244,
301,
218,
59,
280,
283,
55,
499,
133,
316,
407,
146,
56,
394,
386,
297,
285,
137,
58,
214,
142,
6,
124,
48,
60,
212,
75,
50,
412,
458,
127,
45,
266,
202,
368,
138,
260,
41,
193,
88,
114,
410,
95,
382,
416,
281,
434,
359,
98,
462,
300,
352,
230,
247,
117,
64,
287,
405,
224,
19,
259,
305,
220,
150,
477,
111,
448,
78,
103,
7,
385,
151,
429,
325,
273,
317,
470,
454,
170,
223,
5,
307,
396,
315,
53,
154,
446,
24,
255,
227,
76,
456,
250,
321,
330,
391,
355,
49,
479,
387,
216,
39,
251,
312,
217,
136,
262,
322,
344,
466,
242,
100,
388,
38,
323,
376,
379,
279,
239,
85,
306,
181,
485,
120,
333,
334,
17,
395,
81,
374,
147,
139,
185,
42,
1,
424,
199,
225,
113,
438,
128,
338,
156,
493,
46,
160,
11,
3,
171,
464,
62,
238,
431,
440,
302,
65,
308,
348,
125,
174,
195,
77,
392,
249,
82,
350,
444,
232,
186,
494,
384,
275,
129,
294,
246,
357,
102,
96,
73,
15,
263,
296,
236,
29,
340,
152,
149,
143,
437,
172,
190,
34,
158,
254,
295,
483,
397,
337,
72,
343,
178,
404,
270,
346,
205,
377,
486,
497,
370,
414,
240,
360,
490,
94,
256,
8,
54,
398,
183,
228,
162,
399,
289,
83,
86,
197,
243,
57,
25,
288,
488,
372,
168,
206,
188,
491,
452,
353,
478,
421,
221,
430,
184,
204,
26,
211,
140,
155,
468,
161,
420,
303,
30,
449,
131,
500,
20,
71,
79,
445,
425,
293,
411,
400,
320,
474,
272,
413,
329,
177,
122,
21,
347,
314,
451,
101,
367,
311,
40,
476,
415,
418,
363,
282,
469,
89,
274,
481,
475,
203,
268,
393,
261,
200,
121,
164,
472,
10,
284,
14,
358,
153,
383,
67,
473,
373,
191,
144,
16,
345,
361,
433,
116,
331,
489,
66,
106,
487,
426,
99,
27,
141,
264,
439,
371,
213,
18,
253,
292,
130,
409,
278,
419,
90,
496,
447,
465,
461,
339,
80,
31,
70,
233,
326,
37,
265,
252,
222,
118,
198,
406,
286,
380,
104,
304,
351,
408,
180,
22,
364,
381,
401,
234,
375,
459,
319,
229,
207,
291,
52,
463,
427,
23,
235,
32,
208,
192,
349,
231,
354,
435,
182,
428,
332,
378,
290,
108,
258,
471,
115,
47,
457,
166,
43,
495,
63,
110,
107,
422,
324
]`
func GetDB() *commonMsgDatabase {
config.Config.Mongo.DBAddress = []string{"192.168.44.128:37017"}
config.Config.Mongo.DBTimeout = 60
config.Config.Mongo.DBDatabase = "openIM"
config.Config.Mongo.DBSource = "admin"
config.Config.Mongo.DBUserName = "root"
config.Config.Mongo.DBPassword = "openIM123"
config.Config.Mongo.DBMaxPoolSize = 100
config.Config.Mongo.DBRetainChatRecords = 3650
config.Config.Mongo.ChatRecordsClearTime = "0 2 * * 3"
var arr []int
if err := json.Unmarshal([]byte(s), &arr); err != nil {
mongo, err := unrelation.NewMongo()
if err != nil {
panic(err)
}
sort.Ints(arr)
for i, v := range arr {
fmt.Println(i, v, v == i+1)
if v != i+1 {
panic(fmt.Sprintf("expected %d, got %d", i+1, v))
}
err = mongo.GetDatabase().Client().Ping(context.Background(), nil)
if err != nil {
panic(err)
}
return &commonMsgDatabase{
msgDocDatabase: unrelation.NewMsgMongoDriver(mongo.GetDatabase()),
}
}
func Test_Insert(t *testing.T) {
db := GetDB()
ctx := context.Background()
var arr []*unRelationTb.MsgInfoModel
for i := 0; i < 345; i++ {
arr = append(arr, &unRelationTb.MsgInfoModel{
Msg: &unRelationTb.MsgDataModel{
Seq: int64(i),
Content: fmt.Sprintf("test-%d", i),
},
})
}
if err := db.BatchInsertBlock(ctx, "test", arr, 0); err != nil {
t.Fatal(err)
}
}
func Test_Revoke(t *testing.T) {
db := GetDB()
ctx := context.Background()
var arr []*unRelationTb.MsgInfoModel
for i := 0; i < 456; i++ {
arr = append(arr, &unRelationTb.MsgInfoModel{
Revoke: &unRelationTb.RevokeModel{
UserID: "uid_" + strconv.Itoa(i),
Nickname: "uname_" + strconv.Itoa(i),
Time: time.Now().UnixMilli(),
},
})
}
if err := db.BatchInsertBlock(ctx, "test", arr, 123); err != nil {
t.Fatal(err)
}
}
func Test_Delete(t *testing.T) {
db := GetDB()
ctx := context.Background()
var arr []*unRelationTb.MsgInfoModel
for i := 0; i < 123; i++ {
arr = append(arr, &unRelationTb.MsgInfoModel{
DelList: []string{"uid_1", "uid_2"},
})
}
if err := db.BatchInsertBlock(ctx, "test", arr, 210); err != nil {
t.Fatal(err)
}
}