blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-06-05 07:43:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'openimsdk:main' into main

Browse Source
This commit is contained in:
chao 2024-12-12 17:18:08 +08:00 committed by GitHub
commit d8f671d2d3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 158 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
go.mod
View File

@ -14,8 +14,8 @@ require (
github.com/gorilla/websocket v1.5.1
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/protocol v0.0.72-alpha.61
github.com/openimsdk/tools v0.0.50-alpha.47
github.com/openimsdk/protocol v0.0.72-alpha.63
github.com/openimsdk/tools v0.0.50-alpha.50
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0
github.com/stretchr/testify v1.9.0

8
go.sum
View File

@ -347,10 +347,10 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM=
github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/protocol v0.0.72-alpha.61 h1:RuZR9/Sg3p6Bpb2CKPjPoA2AUmTvHITmhZ3PT/RbWMs=
github.com/openimsdk/protocol v0.0.72-alpha.61/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
github.com/openimsdk/tools v0.0.50-alpha.47 h1:Cfe2va/g6WhLjOoQqZkjrdlEDq1dUsfcQsdUB5oADVA=
github.com/openimsdk/tools v0.0.50-alpha.47/go.mod h1:muCtxguNJv8lFwLei27UASu2Nvg4ERSeN0R4K5tivk0=
github.com/openimsdk/protocol v0.0.72-alpha.63 h1:IyPBibEvwBtTmD8DSrlqcekfEXe74k4+KeeHsgdhGh0=
github.com/openimsdk/protocol v0.0.72-alpha.63/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
github.com/openimsdk/tools v0.0.50-alpha.50 h1:+naDlvHcqJDj2NsCGnQd1LLQOET5IRPbrtmWbM/o7JQ=
github.com/openimsdk/tools v0.0.50-alpha.50/go.mod h1:muCtxguNJv8lFwLei27UASu2Nvg4ERSeN0R4K5tivk0=
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

2
internal/api/msg.go
View File

@ -173,6 +173,8 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
data = apistruct.AtElem{}
case constant.Custom:
data = apistruct.CustomElem{}
case constant.Quote:
data = apistruct.QuoteElem{}
case constant.Stream:
data = apistruct.StreamMsgElem{}
case constant.OANotification:

3
internal/msggateway/client.go
View File

@ -18,7 +18,6 @@ import (
"context"
"encoding/json"
"fmt"
"github.com/openimsdk/tools/mw"
"runtime/debug"
"sync"
"sync/atomic"
@ -378,7 +377,7 @@ func (c *Client) activeHeartbeat(ctx context.Context) {
go func() {
defer func() {
if r := recover(); r != nil {
mw.PanicStackToLog(ctx, r)
log.ZPanic(ctx, "activeHeartbeat Panic", r)
}
}()
log.ZDebug(ctx, "server initiative send heartbeat start.")

2
internal/msgtransfer/init.go
View File

@ -204,7 +204,7 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
go func() {
defer func() {
if r := recover(); r != nil {
mw.PanicStackToLog(m.ctx, r)
log.ZPanic(m.ctx, "MsgTransfer Start Panic", r)
}
}()
if err := prommetrics.TransferInit(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {

6
internal/msgtransfer/online_history_msg_handler.go
View File

@ -18,13 +18,13 @@ import (
"context"
"encoding/json"
"errors"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
"github.com/openimsdk/tools/mw"
"strconv"
"strings"
"sync"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
"github.com/IBM/sarama"
"github.com/go-redis/redis"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@ -349,7 +349,7 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
func (och *OnlineHistoryRedisConsumerHandler) HandleUserHasReadSeqMessages(ctx context.Context) {
defer func() {
if r := recover(); r != nil {
mw.PanicStackToLog(ctx, r)
log.ZPanic(ctx, "HandleUserHasReadSeqMessages Panic", r)
}
}()

3
internal/rpc/msg/send.go
View File

@ -16,7 +16,6 @@ package msg
import (
"context"
"github.com/openimsdk/tools/mw"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
@ -89,7 +88,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
defer func() {
if r := recover(); r != nil {
mw.PanicStackToLog(nctx, r)
log.ZPanic(nctx, "setConversationAtInfo Panic", r)
}
}()

4
internal/rpc/msg/sync_msg.go
View File

@ -92,11 +92,13 @@ func (m *msgServer) GetSeqMessage(ctx context.Context, req *msg.GetSeqMessageReq
NotificationMsgs: make(map[string]*sdkws.PullMsgs),
}
for _, conv := range req.Conversations {
_, _, msgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, req.UserID, conv.ConversationID, conv.Seqs)
isEnd, endSeq, msgs, err := m.MsgDatabase.GetMessagesBySeqWithBounds(ctx, req.UserID, conv.ConversationID, conv.Seqs, req.GetOrder())
if err != nil {
return nil, err
}
var pullMsgs *sdkws.PullMsgs
pullMsgs.IsEnd = isEnd
pullMsgs.EndSeq = endSeq
if ok := false; conversationutil.IsNotificationConversationID(conv.ConversationID) {
pullMsgs, ok = resp.NotificationMsgs[conv.ConversationID]
if !ok {

16
internal/rpc/third/log.go
View File

@ -50,14 +50,14 @@ func (t *thirdServer) UploadLogs(ctx context.Context, req *third.UploadLogsReq)
platform := constant.PlatformID2Name[int(req.Platform)]
for _, fileURL := range req.FileURLs {
log := relationtb.Log{
Platform: platform,
UserID: userID,
CreateTime: time.Now(),
Url: fileURL.URL,
FileName: fileURL.Filename,
SystemType: req.AppFramework,
Version: req.Version,
Ex: req.Ex,
Platform: platform,
UserID: userID,
CreateTime: time.Now(),
Url: fileURL.URL,
FileName: fileURL.Filename,
AppFramework: req.AppFramework,
Version: req.Version,
Ex: req.Ex,
}
for i := 0; i < 20; i++ {
id := genLogID()

43
pkg/apistruct/msg.go
View File

@ -14,6 +14,8 @@
package apistruct
import "github.com/openimsdk/protocol/sdkws"
type PictureBaseInfo struct {
UUID string `mapstructure:"uuid"`
Type string `mapstructure:"type" validate:"required"`
@ -90,6 +92,11 @@ type RevokeElem struct {
RevokeMsgClientID string `mapstructure:"revokeMsgClientID" validate:"required"`
}
type QuoteElem struct {
Text string `json:"text,omitempty"`
QuoteMessage *MsgStruct `json:"quoteMessage,omitempty"`
}
type OANotificationElem struct {
NotificationName string `mapstructure:"notificationName" json:"notificationName" validate:"required"`
NotificationFaceURL string `mapstructure:"notificationFaceURL" json:"notificationFaceURL"`
@ -103,6 +110,7 @@ type OANotificationElem struct {
FileElem *FileElem `mapstructure:"fileElem" json:"fileElem"`
Ex string `mapstructure:"ex" json:"ex"`
}
type MessageRevoked struct {
RevokerID string `mapstructure:"revokerID" json:"revokerID" validate:"required"`
RevokerRole int32 `mapstructure:"revokerRole" json:"revokerRole" validate:"required"`
@ -111,3 +119,38 @@ type MessageRevoked struct {
SessionType int32 `mapstructure:"sessionType" json:"sessionType" validate:"required"`
Seq uint32 `mapstructure:"seq" json:"seq" validate:"required"`
}
type MsgStruct struct {
ClientMsgID string `json:"clientMsgID,omitempty"`
ServerMsgID string `json:"serverMsgID,omitempty"`
CreateTime int64 `json:"createTime"`
SendTime int64 `json:"sendTime"`
SessionType int32 `json:"sessionType"`
SendID string `json:"sendID,omitempty"`
RecvID string `json:"recvID,omitempty"`
MsgFrom int32 `json:"msgFrom"`
ContentType int32 `json:"contentType"`
SenderPlatformID int32 `json:"senderPlatformID"`
SenderNickname string `json:"senderNickname,omitempty"`
SenderFaceURL string `json:"senderFaceUrl,omitempty"`
GroupID string `json:"groupID,omitempty"`
Content string `json:"content,omitempty"`
Seq int64 `json:"seq"`
IsRead bool `json:"isRead"`
Status int32 `json:"status"`
IsReact bool `json:"isReact,omitempty"`
IsExternalExtensions bool `json:"isExternalExtensions,omitempty"`
OfflinePush *sdkws.OfflinePushInfo `json:"offlinePush,omitempty"`
AttachedInfo string `json:"attachedInfo,omitempty"`
Ex string `json:"ex,omitempty"`
LocalEx string `json:"localEx,omitempty"`
TextElem *TextElem `json:"textElem,omitempty"`
PictureElem *PictureElem `json:"pictureElem,omitempty"`
SoundElem *SoundElem `json:"soundElem,omitempty"`
VideoElem *VideoElem `json:"videoElem,omitempty"`
FileElem *FileElem `json:"fileElem,omitempty"`
AtTextElem *AtElem `json:"atTextElem,omitempty"`
LocationElem *LocationElem `json:"locationElem,omitempty"`
CustomElem *CustomElem `json:"customElem,omitempty"`
QuoteElem *QuoteElem `json:"quoteElem,omitempty"`
}

81
pkg/common/storage/controller/msg.go
View File

@ -24,6 +24,9 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/redis/go-redis/v9"
"go.mongodb.org/mongo-driver/mongo"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
@ -35,8 +38,6 @@ import (
"github.com/openimsdk/tools/mq/kafka"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/timeutil"
"github.com/redis/go-redis/v9"
"go.mongodb.org/mongo-driver/mongo"
)
const (
@ -56,6 +57,7 @@ type CommonMsgDatabase interface {
GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
// DeleteConversationMsgsAndSetMinSeq deletes conversation messages and resets the minimum sequence number. If `remainTime` is 0, all messages are deleted (this method does not delete Redis
// cache).
GetMessagesBySeqWithBounds(ctx context.Context, userID string, conversationID string, seqs []int64, pullOrder sdkws.PullOrder) (bool, int64, []*sdkws.MsgData, error)
DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
// ClearUserMsgs marks messages for deletion based on clear time and returns a list of sequence numbers for marked messages.
ClearUserMsgs(ctx context.Context, userID string, conversationID string, clearTime int64, lastMsgClearTime time.Time) (seqs []int64, err error)
@ -517,6 +519,81 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
return minSeq, maxSeq, successMsgs, nil
}
func (db *commonMsgDatabase) GetMessagesBySeqWithBounds(ctx context.Context, userID string, conversationID string, seqs []int64, pullOrder sdkws.PullOrder) (bool, int64, []*sdkws.MsgData, error) {
var endSeq int64
var isEnd bool
userMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID)
if err != nil {
return false, 0, nil, err
}
minSeq, err := db.seqConversation.GetMinSeq(ctx, conversationID)
if err != nil {
return false, 0, nil, err
}
maxSeq, err := db.seqConversation.GetMaxSeq(ctx, conversationID)
if err != nil {
return false, 0, nil, err
}
userMaxSeq, err := db.seqUser.GetUserMaxSeq(ctx, conversationID, userID)
if err != nil {
return false, 0, nil, err
}
if userMinSeq > minSeq {
minSeq = userMinSeq
}
if userMaxSeq > 0 && userMaxSeq < maxSeq {
maxSeq = userMaxSeq
}
newSeqs := make([]int64, 0, len(seqs))
for _, seq := range seqs {
if seq <= 0 {
continue
}
// The normal range and can fetch messages
if seq >= minSeq && seq <= maxSeq {
newSeqs = append(newSeqs, seq)
continue
}
// If the requested seq is smaller than the minimum seq and the pull order is descending (pulling older messages)
if seq < minSeq && pullOrder == sdkws.PullOrder_PullOrderDesc {
isEnd = true
endSeq = minSeq
}
// If the requested seq is larger than the maximum seq and the pull order is ascending (pulling newer messages)
if seq > maxSeq && pullOrder == sdkws.PullOrder_PullOrderAsc {
isEnd = true
endSeq = maxSeq
}
}
if len(newSeqs) == 0 {
return isEnd, endSeq, nil, nil
}
successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
if err != nil {
if !errors.Is(err, redis.Nil) {
log.ZWarn(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
}
}
log.ZDebug(ctx, "db.seq.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs",
seqs, "len(successMsgs)", len(successMsgs), "failedSeqs", failedSeqs)
if len(failedSeqs) > 0 {
mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
if err != nil {
return false, 0, nil, err
}
successMsgs = append(successMsgs, mongoMsgs...)
//_, err = db.msg.SetMessagesToCache(ctx, conversationID, mongoMsgs)
//if err != nil {
// return 0, 0, nil, err
//}
}
return isEnd, endSeq, successMsgs, nil
}
func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
var delStruct delMsgRecursionStruct
var skip int64

19
pkg/common/storage/model/log.go
View File

@ -19,13 +19,14 @@ import (
)
type Log struct {
LogID string `bson:"log_id"`
Platform string `bson:"platform"`
UserID string `bson:"user_id"`
CreateTime time.Time `bson:"create_time"`
Url string `bson:"url"`
FileName string `bson:"file_name"`
SystemType string `bson:"system_type"`
Version string `bson:"version"`
Ex string `bson:"ex"`
LogID string `bson:"log_id"`
Platform string `bson:"platform"`
UserID string `bson:"user_id"`
CreateTime time.Time `bson:"create_time"`
Url string `bson:"url"`
FileName string `bson:"file_name"`
SystemType string `bson:"system_type"`
AppFramework string `bson:"app_framework"`
Version string `bson:"version"`
Ex string `bson:"ex"`
}

3
pkg/rpccache/subscriber.go
View File

@ -17,7 +17,6 @@ package rpccache
import (
"context"
"encoding/json"
"github.com/openimsdk/tools/mw"
"github.com/openimsdk/tools/log"
"github.com/redis/go-redis/v9"
@ -26,7 +25,7 @@ import (
func subscriberRedisDeleteCache(ctx context.Context, client redis.UniversalClient, channel string, del func(ctx context.Context, key ...string)) {
defer func() {
if r := recover(); r != nil {
mw.PanicStackToLog(ctx, r)
log.ZPanic(ctx, "subscriberRedisDeleteCache Panic", r)
}
}()
for message := range client.Subscribe(ctx, channel).Channel() {