fix: reserve seq for p2p system messages
This commit is contained in:
+7
-1
@@ -10726,11 +10726,17 @@ func publishIMConversationSystemNotice(tenantID types.TenantID, topic, msgType,
|
|||||||
},
|
},
|
||||||
Content: content,
|
Content: content,
|
||||||
}
|
}
|
||||||
_, err = store.Messages.Save(tenantID, msg, nil, false)
|
saved, err := saveMessageMaybeKafka(context.Background(), tenantID, msg, nil, false)
|
||||||
if err != nil && !errors.Is(err, types.ErrDuplicate) {
|
if err != nil && !errors.Is(err, types.ErrDuplicate) {
|
||||||
logs.Warn.Printf("im system message: failed to save %s for topic %s: %v", template, topic, err)
|
logs.Warn.Printf("im system message: failed to save %s for topic %s: %v", template, topic, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if saved != nil && saved.SeqId > 0 {
|
||||||
|
msg.SeqId = saved.SeqId
|
||||||
|
}
|
||||||
|
if saved != nil && saved.Duplicate {
|
||||||
|
return
|
||||||
|
}
|
||||||
broadcastIMConversationSystemMessage(tenantID, msg, now)
|
broadcastIMConversationSystemMessage(tenantID, msg, now)
|
||||||
sendPushForStoredGroupMessage(tenantID, msg)
|
sendPushForStoredGroupMessage(tenantID, msg)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3397,8 +3397,9 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
|
|||||||
msg.InitTimes()
|
msg.InitTimes()
|
||||||
fromUid := types.ParseUid(msg.From)
|
fromUid := types.ParseUid(msg.From)
|
||||||
isGroupMessage := types.GetTopicCat(msg.Topic) == types.TopicCatGrp
|
isGroupMessage := types.GetTopicCat(msg.Topic) == types.TopicCatGrp
|
||||||
if isGroupMessage && groupMsgRuntime != nil {
|
if groupMsgRuntime != nil {
|
||||||
if clientMsgID != "" {
|
if clientMsgID != "" {
|
||||||
|
if isGroupMessage {
|
||||||
claim, claimed, err := groupMsgRuntime.Claim(tenantID, msg.Topic, fromUid, clientMsgID)
|
claim, claimed, err := groupMsgRuntime.Claim(tenantID, msg.Topic, fromUid, clientMsgID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -3411,6 +3412,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
|
|||||||
return nil, types.ErrDuplicate
|
return nil, types.ErrDuplicate
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
seq, shardMonth, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1, msg.CreatedAt.UTC().Format("200601"))
|
seq, shardMonth, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1, msg.CreatedAt.UTC().Format("200601"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if clientMsgID != "" {
|
if clientMsgID != "" {
|
||||||
@@ -3419,8 +3421,10 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
msg.SeqId = seq
|
msg.SeqId = seq
|
||||||
|
if isGroupMessage {
|
||||||
msg.ShardMonth = shardMonth
|
msg.ShardMonth = shardMonth
|
||||||
}
|
}
|
||||||
|
}
|
||||||
if isGroupMessage && msg.ShardMonth == "" {
|
if isGroupMessage && msg.ShardMonth == "" {
|
||||||
msg.ShardMonth = msg.CreatedAt.UTC().Format("200601")
|
msg.ShardMonth = msg.CreatedAt.UTC().Format("200601")
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user