@ -19,10 +19,11 @@ import (
"context"
"context"
"errors"
"errors"
"go.mongodb.org/mongo-driver/mongo"
pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"go.mongodb.org/mongo-driver/mongo"
)
)
const (
const (
@ -41,12 +42,26 @@ type CommonMsgDatabase interface {
DeleteMessagesFromCache ( ctx context . Context , conversationID string , seqs [ ] int64 ) error
DeleteMessagesFromCache ( ctx context . Context , conversationID string , seqs [ ] int64 ) error
DelUserDeleteMsgsList ( ctx context . Context , conversationID string , seqs [ ] int64 )
DelUserDeleteMsgsList ( ctx context . Context , conversationID string , seqs [ ] int64 )
// incrSeq然后批量插入缓存
// incrSeq然后批量插入缓存
BatchInsertChat2Cache ( ctx context . Context , conversationID string , msgs [ ] * sdkws . MsgData ) ( seq int64 , isNewConversation bool , err error )
BatchInsertChat2Cache (
ctx context . Context ,
conversationID string ,
msgs [ ] * sdkws . MsgData ,
) ( seq int64 , isNewConversation bool , err error )
// 通过seqList获取mongo中写扩散消息
// 通过seqList获取mongo中写扩散消息
GetMsgBySeqsRange ( ctx context . Context , userID string , conversationID string , begin , end , num , userMaxSeq int64 ) ( minSeq int64 , maxSeq int64 , seqMsg [ ] * sdkws . MsgData , err error )
GetMsgBySeqsRange (
ctx context . Context ,
userID string ,
conversationID string ,
begin , end , num , userMaxSeq int64 ,
) ( minSeq int64 , maxSeq int64 , seqMsg [ ] * sdkws . MsgData , err error )
// 通过seqList获取大群在 mongo里面的消息
// 通过seqList获取大群在 mongo里面的消息
GetMsgBySeqs ( ctx context . Context , userID string , conversationID string , seqs [ ] int64 ) ( minSeq int64 , maxSeq int64 , seqMsg [ ] * sdkws . MsgData , err error )
GetMsgBySeqs (
ctx context . Context ,
userID string ,
conversationID string ,
seqs [ ] int64 ,
) ( minSeq int64 , maxSeq int64 , seqMsg [ ] * sdkws . MsgData , err error )
// 删除会话消息重置最小seq, remainTime为消息保留的时间单位秒,超时消息删除, 传0删除所有消息(此方法不删除redis cache)
// 删除会话消息重置最小seq, remainTime为消息保留的时间单位秒,超时消息删除, 传0删除所有消息(此方法不删除redis cache)
DeleteConversationMsgsAndSetMinSeq ( ctx context . Context , conversationID string , remainTime int64 ) error
DeleteConversationMsgsAndSetMinSeq ( ctx context . Context , conversationID string , remainTime int64 ) error
// 用户根据seq删除消息
// 用户根据seq删除消息
@ -73,7 +88,10 @@ type CommonMsgDatabase interface {
UserSetHasReadSeqs ( ctx context . Context , userID string , hasReadSeqs map [ string ] int64 ) error
UserSetHasReadSeqs ( ctx context . Context , userID string , hasReadSeqs map [ string ] int64 ) error
GetMongoMaxAndMinSeq ( ctx context . Context , conversationID string ) ( maxSeq , minSeq int64 , err error )
GetMongoMaxAndMinSeq ( ctx context . Context , conversationID string ) ( maxSeq , minSeq int64 , err error )
GetConversationMinMaxSeqInMongoAndCache ( ctx context . Context , conversationID string ) ( minSeqMongo , maxSeqMongo , minSeqCache , maxSeqCache int64 , err error )
GetConversationMinMaxSeqInMongoAndCache (
ctx context . Context ,
conversationID string ,
) ( minSeqMongo , maxSeqMongo , minSeqCache , maxSeqCache int64 , err error )
SetSendMsgStatus ( ctx context . Context , id string , status int32 ) error
SetSendMsgStatus ( ctx context . Context , id string , status int32 ) error
GetSendMsgStatus ( ctx context . Context , id string ) ( int32 , error )
GetSendMsgStatus ( ctx context . Context , id string ) ( int32 , error )
@ -86,13 +104,38 @@ type CommonMsgDatabase interface {
// modify
// modify
JudgeMessageReactionExist ( ctx context . Context , clientMsgID string , sessionType int32 ) ( bool , error )
JudgeMessageReactionExist ( ctx context . Context , clientMsgID string , sessionType int32 ) ( bool , error )
SetMessageTypeKeyValue ( ctx context . Context , clientMsgID string , sessionType int32 , typeKey , value string ) error
SetMessageTypeKeyValue ( ctx context . Context , clientMsgID string , sessionType int32 , typeKey , value string ) error
SetMessageReactionExpire ( ctx context . Context , clientMsgID string , sessionType int32 , expiration time . Duration ) ( bool , error )
SetMessageReactionExpire (
GetExtendMsg ( ctx context . Context , conversationID string , sessionType int32 , clientMsgID string , maxMsgUpdateTime int64 ) ( * pbMsg . ExtendMsg , error )
ctx context . Context ,
InsertOrUpdateReactionExtendMsgSet ( ctx context . Context , conversationID string , sessionType int32 , clientMsgID string , msgFirstModifyTime int64 , reactionExtensionList map [ string ] * sdkws . KeyValue ) error
clientMsgID string ,
sessionType int32 ,
expiration time . Duration ,
) ( bool , error )
GetExtendMsg (
ctx context . Context ,
conversationID string ,
sessionType int32 ,
clientMsgID string ,
maxMsgUpdateTime int64 ,
) ( * pbMsg . ExtendMsg , error )
InsertOrUpdateReactionExtendMsgSet (
ctx context . Context ,
conversationID string ,
sessionType int32 ,
clientMsgID string ,
msgFirstModifyTime int64 ,
reactionExtensionList map [ string ] * sdkws . KeyValue ,
) error
GetMessageTypeKeyValue ( ctx context . Context , clientMsgID string , sessionType int32 , typeKey string ) ( string , error )
GetMessageTypeKeyValue ( ctx context . Context , clientMsgID string , sessionType int32 , typeKey string ) ( string , error )
GetOneMessageAllReactionList ( ctx context . Context , clientMsgID string , sessionType int32 ) ( map [ string ] string , error )
GetOneMessageAllReactionList ( ctx context . Context , clientMsgID string , sessionType int32 ) ( map [ string ] string , error )
DeleteOneMessageKey ( ctx context . Context , clientMsgID string , sessionType int32 , subKey string ) error
DeleteOneMessageKey ( ctx context . Context , clientMsgID string , sessionType int32 , subKey string ) error
DeleteReactionExtendMsgSet ( ctx context . Context , conversationID string , sessionType int32 , clientMsgID string , msgFirstModifyTime int64 , reactionExtensionList map [ string ] * sdkws . KeyValue ) error
DeleteReactionExtendMsgSet (
ctx context . Context ,
conversationID string ,
sessionType int32 ,
clientMsgID string ,
msgFirstModifyTime int64 ,
reactionExtensionList map [ string ] * sdkws . KeyValue ,
) error
}
}
func NewCommonMsgDatabase ( msgDocModel unRelationTb . MsgDocModelInterface , cacheModel cache . MsgModel ) CommonMsgDatabase {
func NewCommonMsgDatabase ( msgDocModel unRelationTb . MsgDocModelInterface , cacheModel cache . MsgModel ) CommonMsgDatabase {
@ -130,16 +173,32 @@ func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sd
return err
return err
}
}
func ( db * commonMsgDatabase ) MsgToModifyMQ ( ctx context . Context , key , conversationID string , messages [ ] * sdkws . MsgData ) error {
func ( db * commonMsgDatabase ) MsgToModifyMQ (
ctx context . Context ,
key , conversationID string ,
messages [ ] * sdkws . MsgData ,
) error {
if len ( messages ) > 0 {
if len ( messages ) > 0 {
_ , _ , err := db . producerToModify . SendMessage ( ctx , key , & pbMsg . MsgDataToModifyByMQ { ConversationID : conversationID , Messages : messages } )
_ , _ , err := db . producerToModify . SendMessage (
ctx ,
key ,
& pbMsg . MsgDataToModifyByMQ { ConversationID : conversationID , Messages : messages } ,
)
return err
return err
}
}
return nil
return nil
}
}
func ( db * commonMsgDatabase ) MsgToPushMQ ( ctx context . Context , key , conversationID string , msg2mq * sdkws . MsgData ) ( int32 , int64 , error ) {
func ( db * commonMsgDatabase ) MsgToPushMQ (
partition , offset , err := db . producerToPush . SendMessage ( ctx , key , & pbMsg . PushMsgDataToMQ { MsgData : msg2mq , ConversationID : conversationID } )
ctx context . Context ,
key , conversationID string ,
msg2mq * sdkws . MsgData ,
) ( int32 , int64 , error ) {
partition , offset , err := db . producerToPush . SendMessage (
ctx ,
key ,
& pbMsg . PushMsgDataToMQ { MsgData : msg2mq , ConversationID : conversationID } ,
)
if err != nil {
if err != nil {
log . ZError ( ctx , "MsgToPushMQ" , err , "key" , key , "msg2mq" , msg2mq )
log . ZError ( ctx , "MsgToPushMQ" , err , "key" , key , "msg2mq" , msg2mq )
return 0 , 0 , err
return 0 , 0 , err
@ -147,15 +206,30 @@ func (db *commonMsgDatabase) MsgToPushMQ(ctx context.Context, key, conversationI
return partition , offset , nil
return partition , offset , nil
}
}
func ( db * commonMsgDatabase ) MsgToMongoMQ ( ctx context . Context , key , conversationID string , messages [ ] * sdkws . MsgData , lastSeq int64 ) error {
func ( db * commonMsgDatabase ) MsgToMongoMQ (
ctx context . Context ,
key , conversationID string ,
messages [ ] * sdkws . MsgData ,
lastSeq int64 ,
) error {
if len ( messages ) > 0 {
if len ( messages ) > 0 {
_ , _ , err := db . producerToMongo . SendMessage ( ctx , key , & pbMsg . MsgDataToMongoByMQ { LastSeq : lastSeq , ConversationID : conversationID , MsgData : messages } )
_ , _ , err := db . producerToMongo . SendMessage (
ctx ,
key ,
& pbMsg . MsgDataToMongoByMQ { LastSeq : lastSeq , ConversationID : conversationID , MsgData : messages } ,
)
return err
return err
}
}
return nil
return nil
}
}
func ( db * commonMsgDatabase ) BatchInsertBlock ( ctx context . Context , conversationID string , fields [ ] any , key int8 , firstSeq int64 ) error {
func ( db * commonMsgDatabase ) BatchInsertBlock (
ctx context . Context ,
conversationID string ,
fields [ ] any ,
key int8 ,
firstSeq int64 ,
) error {
if len ( fields ) == 0 {
if len ( fields ) == 0 {
return nil
return nil
}
}
@ -256,7 +330,12 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
return nil
return nil
}
}
func ( db * commonMsgDatabase ) BatchInsertChat2DB ( ctx context . Context , conversationID string , msgList [ ] * sdkws . MsgData , currentMaxSeq int64 ) error {
func ( db * commonMsgDatabase ) BatchInsertChat2DB (
ctx context . Context ,
conversationID string ,
msgList [ ] * sdkws . MsgData ,
currentMaxSeq int64 ,
) error {
if len ( msgList ) == 0 {
if len ( msgList ) == 0 {
return errs . ErrArgs . Wrap ( "msgList is empty" )
return errs . ErrArgs . Wrap ( "msgList is empty" )
}
}
@ -302,11 +381,21 @@ func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversatio
return db . BatchInsertBlock ( ctx , conversationID , msgs , updateKeyMsg , msgList [ 0 ] . Seq )
return db . BatchInsertBlock ( ctx , conversationID , msgs , updateKeyMsg , msgList [ 0 ] . Seq )
}
}
func ( db * commonMsgDatabase ) RevokeMsg ( ctx context . Context , conversationID string , seq int64 , revoke * unRelationTb . RevokeModel ) error {
func ( db * commonMsgDatabase ) RevokeMsg (
ctx context . Context ,
conversationID string ,
seq int64 ,
revoke * unRelationTb . RevokeModel ,
) error {
return db . BatchInsertBlock ( ctx , conversationID , [ ] any { revoke } , updateKeyRevoke , seq )
return db . BatchInsertBlock ( ctx , conversationID , [ ] any { revoke } , updateKeyRevoke , seq )
}
}
func ( db * commonMsgDatabase ) MarkSingleChatMsgsAsRead ( ctx context . Context , userID string , conversationID string , totalSeqs [ ] int64 ) error {
func ( db * commonMsgDatabase ) MarkSingleChatMsgsAsRead (
ctx context . Context ,
userID string ,
conversationID string ,
totalSeqs [ ] int64 ,
) error {
for docID , seqs := range db . msg . GetDocIDSeqsMap ( conversationID , totalSeqs ) {
for docID , seqs := range db . msg . GetDocIDSeqsMap ( conversationID , totalSeqs ) {
var indexes [ ] int64
var indexes [ ] int64
for _ , seq := range seqs {
for _ , seq := range seqs {
@ -329,7 +418,11 @@ func (db *commonMsgDatabase) DelUserDeleteMsgsList(ctx context.Context, conversa
db . cache . DelUserDeleteMsgsList ( ctx , conversationID , seqs )
db . cache . DelUserDeleteMsgsList ( ctx , conversationID , seqs )
}
}
func ( db * commonMsgDatabase ) BatchInsertChat2Cache ( ctx context . Context , conversationID string , msgs [ ] * sdkws . MsgData ) ( seq int64 , isNew bool , err error ) {
func ( db * commonMsgDatabase ) BatchInsertChat2Cache (
ctx context . Context ,
conversationID string ,
msgs [ ] * sdkws . MsgData ,
) ( seq int64 , isNew bool , err error ) {
currentMaxSeq , err := db . cache . GetMaxSeq ( ctx , conversationID )
currentMaxSeq , err := db . cache . GetMaxSeq ( ctx , conversationID )
if err != nil && errs . Unwrap ( err ) != redis . Nil {
if err != nil && errs . Unwrap ( err ) != redis . Nil {
prome . Inc ( prome . SeqGetFailedCounter )
prome . Inc ( prome . SeqGetFailedCounter )
@ -376,7 +469,11 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
return lastMaxSeq , isNew , utils . Wrap ( err , "" )
return lastMaxSeq , isNew , utils . Wrap ( err , "" )
}
}
func ( db * commonMsgDatabase ) getMsgBySeqs ( ctx context . Context , userID , conversationID string , seqs [ ] int64 ) ( totalMsgs [ ] * sdkws . MsgData , err error ) {
func ( db * commonMsgDatabase ) getMsgBySeqs (
ctx context . Context ,
userID , conversationID string ,
seqs [ ] int64 ,
) ( totalMsgs [ ] * sdkws . MsgData , err error ) {
for docID , seqs := range db . msg . GetDocIDSeqsMap ( conversationID , seqs ) {
for docID , seqs := range db . msg . GetDocIDSeqsMap ( conversationID , seqs ) {
//log.ZDebug(ctx, "getMsgBySeqs", "docID", docID, "seqs", seqs)
//log.ZDebug(ctx, "getMsgBySeqs", "docID", docID, "seqs", seqs)
msgs , err := db . findMsgInfoBySeq ( ctx , userID , docID , seqs )
msgs , err := db . findMsgInfoBySeq ( ctx , userID , docID , seqs )
@ -390,7 +487,8 @@ func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversat
return totalMsgs , nil
return totalMsgs , nil
}
}
// func (db *commonMsgDatabase) refetchDelSeqsMsgs(ctx context.Context, conversationID string, delNums, rangeBegin, begin int64) (seqMsgs []*unRelationTb.MsgDataModel, err error) {
// func (db *commonMsgDatabase) refetchDelSeqsMsgs(ctx context.Context, conversationID string, delNums, rangeBegin,
// begin int64) (seqMsgs []*unRelationTb.MsgDataModel, err error) {
// var reFetchSeqs []int64
// var reFetchSeqs []int64
// if delNums > 0 {
// if delNums > 0 {
// newBeginSeq := rangeBegin - delNums
// newBeginSeq := rangeBegin - delNums
@ -428,7 +526,11 @@ func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversat
// return seqMsgs, nil
// return seqMsgs, nil
// }
// }
func ( db * commonMsgDatabase ) findMsgInfoBySeq ( ctx context . Context , userID , docID string , seqs [ ] int64 ) ( totalMsgs [ ] * unRelationTb . MsgInfoModel , err error ) {
func ( db * commonMsgDatabase ) findMsgInfoBySeq (
ctx context . Context ,
userID , docID string ,
seqs [ ] int64 ,
) ( totalMsgs [ ] * unRelationTb . MsgInfoModel , err error ) {
msgs , err := db . msgDocDatabase . GetMsgBySeqIndexIn1Doc ( ctx , userID , docID , seqs )
msgs , err := db . msgDocDatabase . GetMsgBySeqIndexIn1Doc ( ctx , userID , docID , seqs )
for _ , msg := range msgs {
for _ , msg := range msgs {
if msg . IsRead {
if msg . IsRead {
@ -438,8 +540,25 @@ func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID
return msgs , err
return msgs , err
}
}
func ( db * commonMsgDatabase ) getMsgBySeqsRange ( ctx context . Context , userID string , conversationID string , allSeqs [ ] int64 , begin , end int64 ) ( seqMsgs [ ] * sdkws . MsgData , err error ) {
func ( db * commonMsgDatabase ) getMsgBySeqsRange (
log . ZDebug ( ctx , "getMsgBySeqsRange" , "conversationID" , conversationID , "allSeqs" , allSeqs , "begin" , begin , "end" , end )
ctx context . Context ,
userID string ,
conversationID string ,
allSeqs [ ] int64 ,
begin , end int64 ,
) ( seqMsgs [ ] * sdkws . MsgData , err error ) {
log . ZDebug (
ctx ,
"getMsgBySeqsRange" ,
"conversationID" ,
conversationID ,
"allSeqs" ,
allSeqs ,
"begin" ,
begin ,
"end" ,
end ,
)
for docID , seqs := range db . msg . GetDocIDSeqsMap ( conversationID , allSeqs ) {
for docID , seqs := range db . msg . GetDocIDSeqsMap ( conversationID , allSeqs ) {
log . ZDebug ( ctx , "getMsgBySeqsRange" , "docID" , docID , "seqs" , seqs )
log . ZDebug ( ctx , "getMsgBySeqsRange" , "docID" , docID , "seqs" , seqs )
msgs , err := db . findMsgInfoBySeq ( ctx , userID , docID , seqs )
msgs , err := db . findMsgInfoBySeq ( ctx , userID , docID , seqs )
@ -456,7 +575,12 @@ func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID strin
return seqMsgs , nil
return seqMsgs , nil
}
}
func ( db * commonMsgDatabase ) GetMsgBySeqsRange ( ctx context . Context , userID string , conversationID string , begin , end , num , userMaxSeq int64 ) ( int64 , int64 , [ ] * sdkws . MsgData , error ) {
func ( db * commonMsgDatabase ) GetMsgBySeqsRange (
ctx context . Context ,
userID string ,
conversationID string ,
begin , end , num , userMaxSeq int64 ,
) ( int64 , int64 , [ ] * sdkws . MsgData , error ) {
userMinSeq , err := db . cache . GetConversationUserMinSeq ( ctx , conversationID , userID )
userMinSeq , err := db . cache . GetConversationUserMinSeq ( ctx , conversationID , userID )
if err != nil && errs . Unwrap ( err ) != redis . Nil {
if err != nil && errs . Unwrap ( err ) != redis . Nil {
return 0 , 0 , nil , err
return 0 , 0 , nil , err
@ -476,7 +600,18 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
if err != nil && errs . Unwrap ( err ) != redis . Nil {
if err != nil && errs . Unwrap ( err ) != redis . Nil {
return 0 , 0 , nil , err
return 0 , 0 , nil , err
}
}
log . ZDebug ( ctx , "GetMsgBySeqsRange" , "userMinSeq" , userMinSeq , "conMinSeq" , minSeq , "conMaxSeq" , maxSeq , "userMaxSeq" , userMaxSeq )
log . ZDebug (
ctx ,
"GetMsgBySeqsRange" ,
"userMinSeq" ,
userMinSeq ,
"conMinSeq" ,
minSeq ,
"conMaxSeq" ,
maxSeq ,
"userMaxSeq" ,
userMaxSeq ,
)
if userMaxSeq != 0 {
if userMaxSeq != 0 {
if userMaxSeq < maxSeq {
if userMaxSeq < maxSeq {
maxSeq = userMaxSeq
maxSeq = userMaxSeq
@ -526,7 +661,18 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
cacheDelNum += 1
cacheDelNum += 1
}
}
}
}
log . ZDebug ( ctx , "get delSeqs from redis" , "delSeqs" , delSeqs , "userID" , userID , "conversationID" , conversationID , "cacheDelNum" , cacheDelNum )
log . ZDebug (
ctx ,
"get delSeqs from redis" ,
"delSeqs" ,
delSeqs ,
"userID" ,
userID ,
"conversationID" ,
conversationID ,
"cacheDelNum" ,
cacheDelNum ,
)
var reGetSeqsCache [ ] int64
var reGetSeqsCache [ ] int64
for i := 1 ; i <= cacheDelNum ; {
for i := 1 ; i <= cacheDelNum ; {
newSeq := newBegin - int64 ( i )
newSeq := newBegin - int64 ( i )
@ -546,7 +692,15 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
if err != nil {
if err != nil {
if err != redis . Nil {
if err != redis . Nil {
prome . Add ( prome . MsgPullFromRedisFailedCounter , len ( failedSeqs2 ) )
prome . Add ( prome . MsgPullFromRedisFailedCounter , len ( failedSeqs2 ) )
log . ZError ( ctx , "get message from redis exception" , err , "conversationID" , conversationID , "seqs" , reGetSeqsCache )
log . ZError (
ctx ,
"get message from redis exception" ,
err ,
"conversationID" ,
conversationID ,
"seqs" ,
reGetSeqsCache ,
)
}
}
}
}
failedSeqs = append ( failedSeqs , failedSeqs2 ... )
failedSeqs = append ( failedSeqs , failedSeqs2 ... )
@ -572,7 +726,12 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
return minSeq , maxSeq , successMsgs , nil
return minSeq , maxSeq , successMsgs , nil
}
}
func ( db * commonMsgDatabase ) GetMsgBySeqs ( ctx context . Context , userID string , conversationID string , seqs [ ] int64 ) ( int64 , int64 , [ ] * sdkws . MsgData , error ) {
func ( db * commonMsgDatabase ) GetMsgBySeqs (
ctx context . Context ,
userID string ,
conversationID string ,
seqs [ ] int64 ,
) ( int64 , int64 , [ ] * sdkws . MsgData , error ) {
userMinSeq , err := db . cache . GetConversationUserMinSeq ( ctx , conversationID , userID )
userMinSeq , err := db . cache . GetConversationUserMinSeq ( ctx , conversationID , userID )
if err != nil && errs . Unwrap ( err ) != redis . Nil {
if err != nil && errs . Unwrap ( err ) != redis . Nil {
return 0 , 0 , nil , err
return 0 , 0 , nil , err
@ -598,10 +757,33 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
if err != nil {
if err != nil {
if err != redis . Nil {
if err != redis . Nil {
prome . Add ( prome . MsgPullFromRedisFailedCounter , len ( failedSeqs ) )
prome . Add ( prome . MsgPullFromRedisFailedCounter , len ( failedSeqs ) )
log . ZError ( ctx , "get message from redis exception" , err , "failedSeqs" , failedSeqs , "conversationID" , conversationID )
log . ZError (
ctx ,
"get message from redis exception" ,
err ,
"failedSeqs" ,
failedSeqs ,
"conversationID" ,
conversationID ,
)
}
}
}
}
log . ZInfo ( ctx , "db.cache.GetMessagesBySeq" , "userID" , userID , "conversationID" , conversationID , "seqs" , seqs , "successMsgs" , len ( successMsgs ) , "failedSeqs" , failedSeqs , "conversationID" , conversationID )
log . ZInfo (
ctx ,
"db.cache.GetMessagesBySeq" ,
"userID" ,
userID ,
"conversationID" ,
conversationID ,
"seqs" ,
seqs ,
"successMsgs" ,
len ( successMsgs ) ,
"failedSeqs" ,
failedSeqs ,
"conversationID" ,
conversationID ,
)
prome . Add ( prome . MsgPullFromRedisSuccessCounter , len ( successMsgs ) )
prome . Add ( prome . MsgPullFromRedisSuccessCounter , len ( successMsgs ) )
if len ( failedSeqs ) > 0 {
if len ( failedSeqs ) > 0 {
mongoMsgs , err := db . getMsgBySeqs ( ctx , userID , conversationID , failedSeqs )
mongoMsgs , err := db . getMsgBySeqs ( ctx , userID , conversationID , failedSeqs )
@ -615,7 +797,11 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
return minSeq , maxSeq , successMsgs , nil
return minSeq , maxSeq , successMsgs , nil
}
}
func ( db * commonMsgDatabase ) DeleteConversationMsgsAndSetMinSeq ( ctx context . Context , conversationID string , remainTime int64 ) error {
func ( db * commonMsgDatabase ) DeleteConversationMsgsAndSetMinSeq (
ctx context . Context ,
conversationID string ,
remainTime int64 ,
) error {
var delStruct delMsgRecursionStruct
var delStruct delMsgRecursionStruct
var skip int64
var skip int64
minSeq , err := db . deleteMsgRecursion ( ctx , conversationID , skip , & delStruct , remainTime )
minSeq , err := db . deleteMsgRecursion ( ctx , conversationID , skip , & delStruct , remainTime )
@ -649,13 +835,26 @@ func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
// seq 70
// seq 70
// set minSeq 21
// set minSeq 21
// recursion 删除list并且返回设置的最小seq
// recursion 删除list并且返回设置的最小seq
func ( db * commonMsgDatabase ) deleteMsgRecursion ( ctx context . Context , conversationID string , index int64 , delStruct * delMsgRecursionStruct , remainTime int64 ) ( int64 , error ) {
func ( db * commonMsgDatabase ) deleteMsgRecursion (
ctx context . Context ,
conversationID string ,
index int64 ,
delStruct * delMsgRecursionStruct ,
remainTime int64 ,
) ( int64 , error ) {
// find from oldest list
// find from oldest list
msgDocModel , err := db . msgDocDatabase . GetMsgDocModelByIndex ( ctx , conversationID , index , 1 )
msgDocModel , err := db . msgDocDatabase . GetMsgDocModelByIndex ( ctx , conversationID , index , 1 )
if err != nil || msgDocModel . DocID == "" {
if err != nil || msgDocModel . DocID == "" {
if err != nil {
if err != nil {
if err == unrelation . ErrMsgListNotExist {
if err == unrelation . ErrMsgListNotExist {
log . ZDebug ( ctx , "deleteMsgRecursion ErrMsgListNotExist" , "conversationID" , conversationID , "index:" , index )
log . ZDebug (
ctx ,
"deleteMsgRecursion ErrMsgListNotExist" ,
"conversationID" ,
conversationID ,
"index:" ,
index ,
)
} else {
} else {
log . ZError ( ctx , "deleteMsgRecursion GetUserMsgListByIndex failed" , err , "conversationID" , conversationID , "index" , index )
log . ZError ( ctx , "deleteMsgRecursion GetUserMsgListByIndex failed" , err , "conversationID" , conversationID , "index" , index )
}
}
@ -667,11 +866,23 @@ func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversatio
}
}
return delStruct . getSetMinSeq ( ) + 1 , nil
return delStruct . getSetMinSeq ( ) + 1 , nil
}
}
log . ZDebug ( ctx , "doc info" , "conversationID" , conversationID , "index" , index , "docID" , msgDocModel . DocID , "len" , len ( msgDocModel . Msg ) )
log . ZDebug (
ctx ,
"doc info" ,
"conversationID" ,
conversationID ,
"index" ,
index ,
"docID" ,
msgDocModel . DocID ,
"len" ,
len ( msgDocModel . Msg ) ,
)
if int64 ( len ( msgDocModel . Msg ) ) > db . msg . GetSingleGocMsgNum ( ) {
if int64 ( len ( msgDocModel . Msg ) ) > db . msg . GetSingleGocMsgNum ( ) {
log . ZWarn ( ctx , "msgs too large" , nil , "lenth" , len ( msgDocModel . Msg ) , "docID:" , msgDocModel . DocID )
log . ZWarn ( ctx , "msgs too large" , nil , "lenth" , len ( msgDocModel . Msg ) , "docID:" , msgDocModel . DocID )
}
}
if msgDocModel . IsFull ( ) && msgDocModel . Msg [ len ( msgDocModel . Msg ) - 1 ] . Msg . SendTime + ( remainTime * 1000 ) < utils . GetCurrentTimestampByMill ( ) {
if msgDocModel . IsFull ( ) &&
msgDocModel . Msg [ len ( msgDocModel . Msg ) - 1 ] . Msg . SendTime + ( remainTime * 1000 ) < utils . GetCurrentTimestampByMill ( ) {
log . ZDebug ( ctx , "doc is full and all msg is expired" , "docID" , msgDocModel . DocID )
log . ZDebug ( ctx , "doc is full and all msg is expired" , "docID" , msgDocModel . DocID )
delStruct . delDocIDs = append ( delStruct . delDocIDs , msgDocModel . DocID )
delStruct . delDocIDs = append ( delStruct . delDocIDs , msgDocModel . DocID )
delStruct . minSeq = msgDocModel . Msg [ len ( msgDocModel . Msg ) - 1 ] . Msg . Seq
delStruct . minSeq = msgDocModel . Msg [ len ( msgDocModel . Msg ) - 1 ] . Msg . Seq
@ -708,7 +919,11 @@ func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversatio
return seq , err
return seq , err
}
}
func ( db * commonMsgDatabase ) DeleteMsgsPhysicalBySeqs ( ctx context . Context , conversationID string , allSeqs [ ] int64 ) error {
func ( db * commonMsgDatabase ) DeleteMsgsPhysicalBySeqs (
ctx context . Context ,
conversationID string ,
allSeqs [ ] int64 ,
) error {
if err := db . cache . DeleteMessages ( ctx , conversationID , allSeqs ) ; err != nil {
if err := db . cache . DeleteMessages ( ctx , conversationID , allSeqs ) ; err != nil {
return err
return err
}
}
@ -724,7 +939,12 @@ func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conve
return nil
return nil
}
}
func ( db * commonMsgDatabase ) DeleteUserMsgsBySeqs ( ctx context . Context , userID string , conversationID string , seqs [ ] int64 ) error {
func ( db * commonMsgDatabase ) DeleteUserMsgsBySeqs (
ctx context . Context ,
userID string ,
conversationID string ,
seqs [ ] int64 ,
) error {
cachedMsgs , _ , err := db . cache . GetMessagesBySeq ( ctx , conversationID , seqs )
cachedMsgs , _ , err := db . cache . GetMessagesBySeq ( ctx , conversationID , seqs )
if err != nil && errs . Unwrap ( err ) != redis . Nil {
if err != nil && errs . Unwrap ( err ) != redis . Nil {
log . ZWarn ( ctx , "DeleteUserMsgsBySeqs" , err , "conversationID" , conversationID , "seqs" , seqs )
log . ZWarn ( ctx , "DeleteUserMsgsBySeqs" , err , "conversationID" , conversationID , "seqs" , seqs )
@ -793,31 +1013,70 @@ func (db *commonMsgDatabase) GetMinSeqs(ctx context.Context, conversationIDs []s
func ( db * commonMsgDatabase ) GetMinSeq ( ctx context . Context , conversationID string ) ( int64 , error ) {
func ( db * commonMsgDatabase ) GetMinSeq ( ctx context . Context , conversationID string ) ( int64 , error ) {
return db . cache . GetMinSeq ( ctx , conversationID )
return db . cache . GetMinSeq ( ctx , conversationID )
}
}
func ( db * commonMsgDatabase ) GetConversationUserMinSeq ( ctx context . Context , conversationID string , userID string ) ( int64 , error ) {
func ( db * commonMsgDatabase ) GetConversationUserMinSeq (
ctx context . Context ,
conversationID string ,
userID string ,
) ( int64 , error ) {
return db . cache . GetConversationUserMinSeq ( ctx , conversationID , userID )
return db . cache . GetConversationUserMinSeq ( ctx , conversationID , userID )
}
}
func ( db * commonMsgDatabase ) GetConversationUserMinSeqs ( ctx context . Context , conversationID string , userIDs [ ] string ) ( map [ string ] int64 , error ) {
func ( db * commonMsgDatabase ) GetConversationUserMinSeqs (
ctx context . Context ,
conversationID string ,
userIDs [ ] string ,
) ( map [ string ] int64 , error ) {
return db . cache . GetConversationUserMinSeqs ( ctx , conversationID , userIDs )
return db . cache . GetConversationUserMinSeqs ( ctx , conversationID , userIDs )
}
}
func ( db * commonMsgDatabase ) SetConversationUserMinSeq ( ctx context . Context , conversationID string , userID string , minSeq int64 ) error {
func ( db * commonMsgDatabase ) SetConversationUserMinSeq (
ctx context . Context ,
conversationID string ,
userID string ,
minSeq int64 ,
) error {
return db . cache . SetConversationUserMinSeq ( ctx , conversationID , userID , minSeq )
return db . cache . SetConversationUserMinSeq ( ctx , conversationID , userID , minSeq )
}
}
func ( db * commonMsgDatabase ) SetConversationUserMinSeqs ( ctx context . Context , conversationID string , seqs map [ string ] int64 ) ( err error ) {
func ( db * commonMsgDatabase ) SetConversationUserMinSeqs (
ctx context . Context ,
conversationID string ,
seqs map [ string ] int64 ,
) ( err error ) {
return db . cache . SetConversationUserMinSeqs ( ctx , conversationID , seqs )
return db . cache . SetConversationUserMinSeqs ( ctx , conversationID , seqs )
}
}
func ( db * commonMsgDatabase ) SetUserConversationsMinSeqs ( ctx context . Context , userID string , seqs map [ string ] int64 ) error {
func ( db * commonMsgDatabase ) SetUserConversationsMinSeqs (
ctx context . Context ,
userID string ,
seqs map [ string ] int64 ,
) error {
return db . cache . SetUserConversationsMinSeqs ( ctx , userID , seqs )
return db . cache . SetUserConversationsMinSeqs ( ctx , userID , seqs )
}
}
func ( db * commonMsgDatabase ) UserSetHasReadSeqs ( ctx context . Context , userID string , hasReadSeqs map [ string ] int64 ) error {
func ( db * commonMsgDatabase ) UserSetHasReadSeqs (
ctx context . Context ,
userID string ,
hasReadSeqs map [ string ] int64 ,
) error {
return db . cache . UserSetHasReadSeqs ( ctx , userID , hasReadSeqs )
return db . cache . UserSetHasReadSeqs ( ctx , userID , hasReadSeqs )
}
}
func ( db * commonMsgDatabase ) SetHasReadSeq ( ctx context . Context , userID string , conversationID string , hasReadSeq int64 ) error {
func ( db * commonMsgDatabase ) SetHasReadSeq (
ctx context . Context ,
userID string ,
conversationID string ,
hasReadSeq int64 ,
) error {
return db . cache . SetHasReadSeq ( ctx , userID , conversationID , hasReadSeq )
return db . cache . SetHasReadSeq ( ctx , userID , conversationID , hasReadSeq )
}
}
func ( db * commonMsgDatabase ) GetHasReadSeqs ( ctx context . Context , userID string , conversationIDs [ ] string ) ( map [ string ] int64 , error ) {
func ( db * commonMsgDatabase ) GetHasReadSeqs (
ctx context . Context ,
userID string ,
conversationIDs [ ] string ,
) ( map [ string ] int64 , error ) {
return db . cache . GetHasReadSeqs ( ctx , userID , conversationIDs )
return db . cache . GetHasReadSeqs ( ctx , userID , conversationIDs )
}
}
@ -833,7 +1092,10 @@ func (db *commonMsgDatabase) GetSendMsgStatus(ctx context.Context, id string) (i
return db . cache . GetSendMsgStatus ( ctx , id )
return db . cache . GetSendMsgStatus ( ctx , id )
}
}
func ( db * commonMsgDatabase ) GetConversationMinMaxSeqInMongoAndCache ( ctx context . Context , conversationID string ) ( minSeqMongo , maxSeqMongo , minSeqCache , maxSeqCache int64 , err error ) {
func ( db * commonMsgDatabase ) GetConversationMinMaxSeqInMongoAndCache (
ctx context . Context ,
conversationID string ,
) ( minSeqMongo , maxSeqMongo , minSeqCache , maxSeqCache int64 , err error ) {
minSeqMongo , maxSeqMongo , err = db . GetMinMaxSeqMongo ( ctx , conversationID )
minSeqMongo , maxSeqMongo , err = db . GetMinMaxSeqMongo ( ctx , conversationID )
if err != nil {
if err != nil {
return
return
@ -849,11 +1111,17 @@ func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(ctx context
return
return
}
}
func ( db * commonMsgDatabase ) GetMongoMaxAndMinSeq ( ctx context . Context , conversationID string ) ( maxSeq , minSeq int64 , err error ) {
func ( db * commonMsgDatabase ) GetMongoMaxAndMinSeq (
ctx context . Context ,
conversationID string ,
) ( maxSeq , minSeq int64 , err error ) {
return db . GetMinMaxSeqMongo ( ctx , conversationID )
return db . GetMinMaxSeqMongo ( ctx , conversationID )
}
}
func ( db * commonMsgDatabase ) GetMinMaxSeqMongo ( ctx context . Context , conversationID string ) ( minSeqMongo , maxSeqMongo int64 , err error ) {
func ( db * commonMsgDatabase ) GetMinMaxSeqMongo (
ctx context . Context ,
conversationID string ,
) ( minSeqMongo , maxSeqMongo int64 , err error ) {
oldestMsgMongo , err := db . msgDocDatabase . GetOldestMsg ( ctx , conversationID )
oldestMsgMongo , err := db . msgDocDatabase . GetOldestMsg ( ctx , conversationID )
if err != nil {
if err != nil {
return
return
@ -867,35 +1135,83 @@ func (db *commonMsgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversation
return
return
}
}
func ( db * commonMsgDatabase ) JudgeMessageReactionExist ( ctx context . Context , clientMsgID string , sessionType int32 ) ( bool , error ) {
func ( db * commonMsgDatabase ) JudgeMessageReactionExist (
ctx context . Context ,
clientMsgID string ,
sessionType int32 ,
) ( bool , error ) {
return db . cache . JudgeMessageReactionExist ( ctx , clientMsgID , sessionType )
return db . cache . JudgeMessageReactionExist ( ctx , clientMsgID , sessionType )
}
}
func ( db * commonMsgDatabase ) SetMessageTypeKeyValue ( ctx context . Context , clientMsgID string , sessionType int32 , typeKey , value string ) error {
func ( db * commonMsgDatabase ) SetMessageTypeKeyValue (
ctx context . Context ,
clientMsgID string ,
sessionType int32 ,
typeKey , value string ,
) error {
return db . cache . SetMessageTypeKeyValue ( ctx , clientMsgID , sessionType , typeKey , value )
return db . cache . SetMessageTypeKeyValue ( ctx , clientMsgID , sessionType , typeKey , value )
}
}
func ( db * commonMsgDatabase ) SetMessageReactionExpire ( ctx context . Context , clientMsgID string , sessionType int32 , expiration time . Duration ) ( bool , error ) {
func ( db * commonMsgDatabase ) SetMessageReactionExpire (
ctx context . Context ,
clientMsgID string ,
sessionType int32 ,
expiration time . Duration ,
) ( bool , error ) {
return db . cache . SetMessageReactionExpire ( ctx , clientMsgID , sessionType , expiration )
return db . cache . SetMessageReactionExpire ( ctx , clientMsgID , sessionType , expiration )
}
}
func ( db * commonMsgDatabase ) GetMessageTypeKeyValue ( ctx context . Context , clientMsgID string , sessionType int32 , typeKey string ) ( string , error ) {
func ( db * commonMsgDatabase ) GetMessageTypeKeyValue (
ctx context . Context ,
clientMsgID string ,
sessionType int32 ,
typeKey string ,
) ( string , error ) {
return db . cache . GetMessageTypeKeyValue ( ctx , clientMsgID , sessionType , typeKey )
return db . cache . GetMessageTypeKeyValue ( ctx , clientMsgID , sessionType , typeKey )
}
}
func ( db * commonMsgDatabase ) GetOneMessageAllReactionList ( ctx context . Context , clientMsgID string , sessionType int32 ) ( map [ string ] string , error ) {
func ( db * commonMsgDatabase ) GetOneMessageAllReactionList (
ctx context . Context ,
clientMsgID string ,
sessionType int32 ,
) ( map [ string ] string , error ) {
return db . cache . GetOneMessageAllReactionList ( ctx , clientMsgID , sessionType )
return db . cache . GetOneMessageAllReactionList ( ctx , clientMsgID , sessionType )
}
}
func ( db * commonMsgDatabase ) DeleteOneMessageKey ( ctx context . Context , clientMsgID string , sessionType int32 , subKey string ) error {
func ( db * commonMsgDatabase ) DeleteOneMessageKey (
ctx context . Context ,
clientMsgID string ,
sessionType int32 ,
subKey string ,
) error {
return db . cache . DeleteOneMessageKey ( ctx , clientMsgID , sessionType , subKey )
return db . cache . DeleteOneMessageKey ( ctx , clientMsgID , sessionType , subKey )
}
}
func ( db * commonMsgDatabase ) InsertOrUpdateReactionExtendMsgSet ( ctx context . Context , conversationID string , sessionType int32 , clientMsgID string , msgFirstModifyTime int64 , reactionExtensions map [ string ] * sdkws . KeyValue ) error {
func ( db * commonMsgDatabase ) InsertOrUpdateReactionExtendMsgSet (
return db . extendMsgDatabase . InsertOrUpdateReactionExtendMsgSet ( ctx , conversationID , sessionType , clientMsgID , msgFirstModifyTime , db . extendMsgSetModel . Pb2Model ( reactionExtensions ) )
ctx context . Context ,
conversationID string ,
sessionType int32 ,
clientMsgID string ,
msgFirstModifyTime int64 ,
reactionExtensions map [ string ] * sdkws . KeyValue ,
) error {
return db . extendMsgDatabase . InsertOrUpdateReactionExtendMsgSet (
ctx ,
conversationID ,
sessionType ,
clientMsgID ,
msgFirstModifyTime ,
db . extendMsgSetModel . Pb2Model ( reactionExtensions ) ,
)
}
}
func ( db * commonMsgDatabase ) GetExtendMsg ( ctx context . Context , conversationID string , sessionType int32 , clientMsgID string , maxMsgUpdateTime int64 ) ( * pbMsg . ExtendMsg , error ) {
func ( db * commonMsgDatabase ) GetExtendMsg (
ctx context . Context ,
conversationID string ,
sessionType int32 ,
clientMsgID string ,
maxMsgUpdateTime int64 ,
) ( * pbMsg . ExtendMsg , error ) {
extendMsgSet , err := db . extendMsgDatabase . GetExtendMsgSet ( ctx , conversationID , sessionType , maxMsgUpdateTime )
extendMsgSet , err := db . extendMsgDatabase . GetExtendMsgSet ( ctx , conversationID , sessionType , maxMsgUpdateTime )
if err != nil {
if err != nil {
return nil , err
return nil , err
@ -923,6 +1239,20 @@ func (db *commonMsgDatabase) GetExtendMsg(ctx context.Context, conversationID st
} , nil
} , nil
}
}
func ( db * commonMsgDatabase ) DeleteReactionExtendMsgSet ( ctx context . Context , conversationID string , sessionType int32 , clientMsgID string , msgFirstModifyTime int64 , reactionExtensions map [ string ] * sdkws . KeyValue ) error {
func ( db * commonMsgDatabase ) DeleteReactionExtendMsgSet (
return db . extendMsgDatabase . DeleteReactionExtendMsgSet ( ctx , conversationID , sessionType , clientMsgID , msgFirstModifyTime , db . extendMsgSetModel . Pb2Model ( reactionExtensions ) )
ctx context . Context ,
conversationID string ,
sessionType int32 ,
clientMsgID string ,
msgFirstModifyTime int64 ,
reactionExtensions map [ string ] * sdkws . KeyValue ,
) error {
return db . extendMsgDatabase . DeleteReactionExtendMsgSet (
ctx ,
conversationID ,
sessionType ,
clientMsgID ,
msgFirstModifyTime ,
db . extendMsgSetModel . Pb2Model ( reactionExtensions ) ,
)
}
}