package unrelation import ( "Open_IM/pkg/common/config" "Open_IM/pkg/common/constant" "Open_IM/pkg/common/db" "Open_IM/pkg/common/log" promePkg "Open_IM/pkg/common/prometheus" pbMsg "Open_IM/pkg/proto/msg" "Open_IM/pkg/utils" "context" "errors" go_redis "github.com/go-redis/redis/v8" "github.com/golang/protobuf/proto" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" ) func (d *db.DataBases) BatchInsertChat2DB(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string, currentMaxSeq uint64) error { newTime := getCurrentTimestampByMill() if len(msgList) > GetSingleGocMsgNum() { return errors.New("too large") } isInit := false var remain uint64 blk0 := uint64(GetSingleGocMsgNum() - 1) //currentMaxSeq 4998 if currentMaxSeq < uint64(mongo2.GetSingleGocMsgNum()) { remain = blk0 - currentMaxSeq //1 } else { excludeBlk0 := currentMaxSeq - blk0 //=1 //(5000-1)%5000 == 4999 remain = (uint64(mongo2.GetSingleGocMsgNum()) - (excludeBlk0 % uint64(mongo2.GetSingleGocMsgNum()))) % uint64(mongo2.GetSingleGocMsgNum()) } //remain=1 insertCounter := uint64(0) msgListToMongo := make([]mongo2.MsgInfo, 0) msgListToMongoNext := make([]mongo2.MsgInfo, 0) seqUid := "" seqUidNext := "" log.Debug(operationID, "remain ", remain, "insertCounter ", insertCounter, "currentMaxSeq ", currentMaxSeq, userID, len(msgList)) var err error for _, m := range msgList { log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID) currentMaxSeq++ sMsg := mongo2.MsgInfo{} sMsg.SendTime = m.MsgData.SendTime m.MsgData.Seq = uint32(currentMaxSeq) log.Debug(operationID, "mongo msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", userID, "seq: ", currentMaxSeq) if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil { return utils.Wrap(err, "") } if isInit { msgListToMongoNext = append(msgListToMongoNext, sMsg) seqUidNext = mongo2.getSeqUid(userID, uint32(currentMaxSeq)) log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) continue } if insertCounter < remain { msgListToMongo = append(msgListToMongo, sMsg) insertCounter++ seqUid = mongo2.getSeqUid(userID, uint32(currentMaxSeq)) log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID) } else { msgListToMongoNext = append(msgListToMongoNext, sMsg) seqUidNext = mongo2.getSeqUid(userID, uint32(currentMaxSeq)) log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID) } } ctx := context.Background() c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(mongo2.cChat) if seqUid != "" { filter := bson.M{"uid": seqUid} log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo, "userID: ", userID) err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err() if err != nil { if err == mongo.ErrNoDocuments { filter := bson.M{"uid": seqUid} sChat := mongo2.UserChat{} sChat.UID = seqUid sChat.Msg = msgListToMongo log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo) if _, err = c.InsertOne(ctx, &sChat); err != nil { promePkg.PromeInc(promePkg.MsgInsertMongoFailedCounter) log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) return utils.Wrap(err, "") } promePkg.PromeInc(promePkg.MsgInsertMongoSuccessCounter) } else { promePkg.PromeInc(promePkg.MsgInsertMongoFailedCounter) log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter) return utils.Wrap(err, "") } } else { promePkg.PromeInc(promePkg.MsgInsertMongoSuccessCounter) } } if seqUidNext != "" { filter := bson.M{"uid": seqUidNext} sChat := mongo2.UserChat{} sChat.UID = seqUidNext sChat.Msg = msgListToMongoNext log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext, "userID: ", userID) if _, err = c.InsertOne(ctx, &sChat); err != nil { promePkg.PromeInc(promePkg.MsgInsertMongoFailedCounter) log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) return utils.Wrap(err, "") } promePkg.PromeInc(promePkg.MsgInsertMongoSuccessCounter) } log.Debug(operationID, "batch mgo cost time ", mongo2.getCurrentTimestampByMill()-newTime, userID, len(msgList)) return nil } func (d *db.DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) { newTime := mongo2.getCurrentTimestampByMill() lenList := len(msgList) if lenList > mongo2.GetSingleGocMsgNum() { return errors.New("too large"), 0 } if lenList < 1 { return errors.New("too short as 0"), 0 } // judge sessionType to get seq var currentMaxSeq uint64 var err error if msgList[0].MsgData.SessionType == constant.SuperGroupChatType { currentMaxSeq, err = d.GetGroupMaxSeq(insertID) log.Debug(operationID, "constant.SuperGroupChatType lastMaxSeq before add ", currentMaxSeq, "userID ", insertID, err) } else { currentMaxSeq, err = d.GetUserMaxSeq(insertID) log.Debug(operationID, "constant.SingleChatType lastMaxSeq before add ", currentMaxSeq, "userID ", insertID, err) } if err != nil && err != go_redis.Nil { promePkg.PromeInc(promePkg.SeqGetFailedCounter) return utils.Wrap(err, ""), 0 } promePkg.PromeInc(promePkg.SeqGetSuccessCounter) lastMaxSeq := currentMaxSeq for _, m := range msgList { currentMaxSeq++ sMsg := mongo2.MsgInfo{} sMsg.SendTime = m.MsgData.SendTime m.MsgData.Seq = uint32(currentMaxSeq) log.Debug(operationID, "cache msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", insertID, "seq: ", currentMaxSeq) } log.Debug(operationID, "SetMessageToCache ", insertID, len(msgList)) err, failedNum := d.SetMessageToCache(msgList, insertID, operationID) if err != nil { promePkg.PromeAdd(promePkg.MsgInsertRedisFailedCounter, failedNum) log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), insertID) } else { promePkg.PromeInc(promePkg.MsgInsertRedisSuccessCounter) } log.Debug(operationID, "batch to redis cost time ", mongo2.getCurrentTimestampByMill()-newTime, insertID, len(msgList)) if msgList[0].MsgData.SessionType == constant.SuperGroupChatType { err = d.SetGroupMaxSeq(insertID, currentMaxSeq) } else { err = d.SetUserMaxSeq(insertID, currentMaxSeq) } if err != nil { promePkg.PromeInc(promePkg.SeqSetFailedCounter) } else { promePkg.PromeInc(promePkg.SeqSetSuccessCounter) } return utils.Wrap(err, ""), lastMaxSeq }