diff --git a/go.mod b/go.mod index 0c9e3d8c3..58dfc27f0 100644 --- a/go.mod +++ b/go.mod @@ -43,7 +43,7 @@ require ( require github.com/google/uuid v1.3.0 require ( - github.com/OpenIMSDK/openKeeper v0.0.8 + github.com/OpenIMSDK/openKeeper v0.0.9 github.com/go-redis/redis v6.15.9+incompatible github.com/go-sql-driver/mysql v1.6.0 k8s.io/apimachinery v0.27.1 diff --git a/go.sum b/go.sum index 0295203b6..8799074c9 100644 --- a/go.sum +++ b/go.sum @@ -379,8 +379,8 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= -github.com/OpenIMSDK/openKeeper v0.0.8 h1:HUL8ZPsrIA4fJ86TZ+cBbsRnUA1ZFgGs0YiXRpiuM+E= -github.com/OpenIMSDK/openKeeper v0.0.8/go.mod h1:HwHgo+p3E+aAeiLOSyDwVB2zFJtu+QvnFNlfNjFG9vA= +github.com/OpenIMSDK/openKeeper v0.0.9 h1:KzXHvWivCCOa573DKd6L38Usr1NLFT29gqRupkNOZHM= +github.com/OpenIMSDK/openKeeper v0.0.9/go.mod h1:HwHgo+p3E+aAeiLOSyDwVB2zFJtu+QvnFNlfNjFG9vA= github.com/OpenIMSDK/open_utils v1.0.8 h1:IopxWgJwEF5ZAPsRuiZZOfcxNOQOCt/p8VDENcHN9r4= github.com/OpenIMSDK/open_utils v1.0.8/go.mod h1:FLoaQblWUVKQgqt2LrNzfSZLT6D3DICBn1kcOMDLUOI= github.com/Shopify/sarama v1.29.0 h1:ARid8o8oieau9XrHI55f/L3EoRAhm9px6sonbD7yuUE= diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go index 013dd015b..38ad1f376 100644 --- a/internal/msgtransfer/online_history_msg_handler.go +++ b/internal/msgtransfer/online_history_msg_handler.go @@ -20,7 +20,7 @@ import ( "github.com/OpenIMSDK/Open-IM-Server/pkg/utils" "github.com/Shopify/sarama" "github.com/go-redis/redis" - "github.com/golang/protobuf/proto" + "google.golang.org/protobuf/proto" ) const ConsumerMsgs = 3 diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index 8da3d3231..96060461d 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -104,7 +104,6 @@ func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGr for msg := range claim.Messages() { ctx := mc.historyConsumerGroup.GetContextFromMsg(msg) if len(msg.Value) != 0 { - log.ZDebug(ctx, "mongo consumer recv new msg", "conversationID", msg.Key, "offset", msg.Offset) mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess) } else { log.ZError(ctx, "mongo msg get from kafka but is nil", nil, "conversationID", msg.Key) diff --git a/pkg/a2r/api2rpc.go b/pkg/a2r/api2rpc.go index d6b56696d..ca351a13a 100644 --- a/pkg/a2r/api2rpc.go +++ b/pkg/a2r/api2rpc.go @@ -15,6 +15,7 @@ func Call[A, B, C any]( client func(ctx context.Context) (C, error), c *gin.Context, ) { + log.ZDebug(c, "before bind") var req A if err := c.BindJSON(&req); err != nil { log.ZWarn(c, "gin bind json error", err, "req", req) diff --git a/pkg/common/db/controller/common_msg.go b/pkg/common/db/controller/common_msg.go index da4d785cb..08584f2c3 100644 --- a/pkg/common/db/controller/common_msg.go +++ b/pkg/common/db/controller/common_msg.go @@ -26,7 +26,7 @@ import ( "github.com/go-redis/redis/v8" "go.mongodb.org/mongo-driver/mongo" - "github.com/golang/protobuf/proto" + "google.golang.org/protobuf/proto" ) type CommonMsgDatabase interface { diff --git a/pkg/common/db/controller/notification.go b/pkg/common/db/controller/notification.go index 3e70103df..032a0fa51 100644 --- a/pkg/common/db/controller/notification.go +++ b/pkg/common/db/controller/notification.go @@ -25,7 +25,7 @@ import ( "github.com/go-redis/redis/v8" "go.mongodb.org/mongo-driver/mongo" - "github.com/golang/protobuf/proto" + "google.golang.org/protobuf/proto" ) type NotificationDatabase interface { @@ -229,7 +229,6 @@ func (db *notificationDatabase) GetGroupMinSeq(ctx context.Context, groupID stri } func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error { - //newTime := utils.GetCurrentTimestampByMill() if int64(len(msgList)) > db.msg.GetsingleGocNotificationNum() { return errors.New("too large") } @@ -251,7 +250,6 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa docIDNext := "" var err error for _, m := range msgList { - //log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID) currentMaxSeq++ sMsg := unRelationTb.NotificationInfoModel{} sMsg.SendTime = m.SendTime @@ -263,18 +261,13 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa msgsToMongo = append(msgsToMongo, sMsg) insertCounter++ docID = db.msg.GetDocID(conversationID, currentMaxSeq) - //log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID) } else { msgsToMongoNext = append(msgsToMongoNext, sMsg) docIDNext = db.msg.GetDocID(conversationID, currentMaxSeq) - //log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID) } } if docID != "" { - //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": msgsToMongo}}}).Err() err = db.msgDocDatabase.PushMsgsToDoc(ctx, docID, msgsToMongo) if err != nil { if err == mongo.ErrNoDocuments { @@ -283,13 +276,11 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa doc.Msg = msgsToMongo if err = db.msgDocDatabase.Create(ctx, doc); err != nil { prome.Inc(prome.MsgInsertMongoFailedCounter) - //log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) return utils.Wrap(err, "") } prome.Inc(prome.MsgInsertMongoSuccessCounter) } else { prome.Inc(prome.MsgInsertMongoFailedCounter) - //log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter) return utils.Wrap(err, "") } } else { @@ -300,15 +291,12 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa nextDoc := &unRelationTb.NotificationDocModel{} nextDoc.DocID = docIDNext nextDoc.Msg = msgsToMongoNext - //log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext, "userID: ", userID) if err = db.msgDocDatabase.Create(ctx, nextDoc); err != nil { prome.Inc(prome.MsgInsertMongoFailedCounter) - //log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) return utils.Wrap(err, "") } prome.Inc(prome.MsgInsertMongoSuccessCounter) } - //log.Debug(operationID, "batch mgo cost time ", mongo2.getCurrentTimestampByMill()-newTime, userID, len(msgList)) return nil } diff --git a/pkg/common/db/unrelation/extend_msg.go b/pkg/common/db/unrelation/extend_msg.go index 85b97775c..ab2b8b2b3 100644 --- a/pkg/common/db/unrelation/extend_msg.go +++ b/pkg/common/db/unrelation/extend_msg.go @@ -36,7 +36,7 @@ func (e *ExtendMsgSetMongoDriver) GetAllExtendMsgSet(ctx context.Context, ID str findOpts.SetProjection(bson.M{"extend_msgs": 0}) } } - cursor, err := e.ExtendMsgSetCollection.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: regex}}, findOpts) + cursor, err := e.ExtendMsgSetCollection.Find(ctx, bson.M{"doc_id": primitive.Regex{Pattern: regex}}, findOpts) if err != nil { return nil, utils.Wrap(err, "") } diff --git a/pkg/common/db/unrelation/msg.go b/pkg/common/db/unrelation/msg.go index d54f60e25..65b412ea8 100644 --- a/pkg/common/db/unrelation/msg.go +++ b/pkg/common/db/unrelation/msg.go @@ -30,8 +30,7 @@ func NewMsgMongoDriver(database *mongo.Database) table.MsgDocModelInterface { } func (m *MsgMongoDriver) PushMsgsToDoc(ctx context.Context, docID string, msgsToMongo []table.MsgInfoModel) error { - filter := bson.M{"uid": docID} - return m.MsgCollection.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err() + return m.MsgCollection.FindOneAndUpdate(ctx, bson.M{"doc_id": docID}, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err() } func (m *MsgMongoDriver) Create(ctx context.Context, model *table.MsgDocModel) error { @@ -45,7 +44,7 @@ func (m *MsgMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Context, doc if err != nil { return utils.Wrap(err, "") } - _, err = m.MsgCollection.UpdateOne(ctx, bson.M{"uid": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}}) + _, err = m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}}) if err != nil { return utils.Wrap(err, "") } @@ -54,13 +53,13 @@ func (m *MsgMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Context, doc func (m *MsgMongoDriver) FindOneByDocID(ctx context.Context, docID string) (*table.MsgDocModel, error) { doc := &table.MsgDocModel{} - err := m.MsgCollection.FindOne(ctx, bson.M{"uid": docID}).Decode(doc) + err := m.MsgCollection.FindOne(ctx, bson.M{"doc_id": docID}).Decode(doc) return doc, err } func (m *MsgMongoDriver) GetMsgsByIndex(ctx context.Context, conversationID string, index int64) (*table.MsgDocModel, error) { - findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"uid": 1}) - cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts) + findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"doc_id": 1}) + cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts) if err != nil { return nil, utils.Wrap(err, "") } @@ -77,7 +76,7 @@ func (m *MsgMongoDriver) GetMsgsByIndex(ctx context.Context, conversationID stri func (m *MsgMongoDriver) GetNewestMsg(ctx context.Context, conversationID string) (*table.MsgInfoModel, error) { var msgDocs []table.MsgDocModel - cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": -1})) + cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": -1})) if err != nil { return nil, utils.Wrap(err, "") } @@ -96,7 +95,7 @@ func (m *MsgMongoDriver) GetNewestMsg(ctx context.Context, conversationID string func (m *MsgMongoDriver) GetOldestMsg(ctx context.Context, conversationID string) (*table.MsgInfoModel, error) { var msgDocs []table.MsgDocModel - cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": 1})) + cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": 1})) if err != nil { return nil, err } @@ -126,12 +125,12 @@ func (m *MsgMongoDriver) Delete(ctx context.Context, docIDs []string) error { if docIDs == nil { return nil } - _, err := m.MsgCollection.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": docIDs}}) + _, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": bson.M{"$in": docIDs}}) return err } func (m *MsgMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.MsgDocModel) error { - _, err := m.MsgCollection.UpdateOne(ctx, bson.M{"uid": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}}) + _, err := m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}}) return err } diff --git a/pkg/common/db/unrelation/notification.go b/pkg/common/db/unrelation/notification.go index dead67df6..bd539d356 100644 --- a/pkg/common/db/unrelation/notification.go +++ b/pkg/common/db/unrelation/notification.go @@ -28,7 +28,7 @@ func NewNotificationMongoDriver(database *mongo.Database) table.NotificationDocM } func (m *NotificationMongoDriver) PushMsgsToDoc(ctx context.Context, docID string, msgsToMongo []table.NotificationInfoModel) error { - filter := bson.M{"uid": docID} + filter := bson.M{"doc_id": docID} return m.MsgCollection.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err() } @@ -43,7 +43,7 @@ func (m *NotificationMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Con if err != nil { return utils.Wrap(err, "") } - _, err = m.MsgCollection.UpdateOne(ctx, bson.M{"uid": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}}) + _, err = m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}}) if err != nil { return utils.Wrap(err, "") } @@ -52,13 +52,13 @@ func (m *NotificationMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Con func (m *NotificationMongoDriver) FindOneByDocID(ctx context.Context, docID string) (*table.NotificationDocModel, error) { doc := &table.NotificationDocModel{} - err := m.MsgCollection.FindOne(ctx, bson.M{"uid": docID}).Decode(doc) + err := m.MsgCollection.FindOne(ctx, bson.M{"doc_id": docID}).Decode(doc) return doc, err } func (m *NotificationMongoDriver) GetMsgsByIndex(ctx context.Context, conversationID string, index int64) (*table.NotificationDocModel, error) { - findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"uid": 1}) - cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts) + findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"doc_id": 1}) + cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts) if err != nil { return nil, utils.Wrap(err, "") } @@ -75,7 +75,7 @@ func (m *NotificationMongoDriver) GetMsgsByIndex(ctx context.Context, conversati func (m *NotificationMongoDriver) GetNewestMsg(ctx context.Context, conversationID string) (*table.NotificationInfoModel, error) { var msgDocs []table.NotificationDocModel - cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": -1})) + cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": -1})) if err != nil { return nil, utils.Wrap(err, "") } @@ -94,7 +94,7 @@ func (m *NotificationMongoDriver) GetNewestMsg(ctx context.Context, conversation func (m *NotificationMongoDriver) GetOldestMsg(ctx context.Context, conversationID string) (*table.NotificationInfoModel, error) { var msgDocs []table.NotificationDocModel - cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": 1})) + cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": 1})) if err != nil { return nil, err } @@ -124,11 +124,11 @@ func (m *NotificationMongoDriver) Delete(ctx context.Context, docIDs []string) e if docIDs == nil { return nil } - _, err := m.MsgCollection.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": docIDs}}) + _, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": bson.M{"$in": docIDs}}) return err } func (m *NotificationMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.NotificationDocModel) error { - _, err := m.MsgCollection.UpdateOne(ctx, bson.M{"uid": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}}) + _, err := m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}}) return err } diff --git a/pkg/common/kafka/producer.go b/pkg/common/kafka/producer.go index 6c6d6f149..da2dd9155 100644 --- a/pkg/common/kafka/producer.go +++ b/pkg/common/kafka/producer.go @@ -4,7 +4,6 @@ import ( "context" "errors" "strconv" - "strings" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/config" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant" @@ -93,7 +92,6 @@ func (p *Producer) SendMessage(ctx context.Context, key string, m proto.Message) for i, header := range header { arr = append(arr, strconv.Itoa(i), string(header.Key), string(header.Value)) } - log.ZInfo(ctx, "producer.kafka.GetContextWithMQHeader", "len", len(header), "header", strings.Join(arr, ", ")) kMsg.Headers = header partition, offset, err := p.producer.SendMessage(kMsg) log.ZDebug(ctx, "ByteEncoder SendMessage end", "key ", kMsg.Key, "key length", kMsg.Value.Length())