Merge remote-tracking branch 'origin/main' into js-server

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

* go.mod

* fix: nil pointer error on close

* fix: listen error

* fix: listen error

* update go.mod

* feat: add log

* fix: token parse token value

* fix: GetMsgBySeqs boundary issues

* fix: sn_ not sort

* fix: sn_ not sort

* fix: sn_ not sort

* fix: jssdk add

* fix: jssdk support

* fix: jssdk support

* fix: jssdk support

* fix: the message I sent is not set to read seq in mongodb

* fix: cannot modify group member avatars

* fix: MemberEnterNotification

* fix: MemberEnterNotification

* fix: MsgData status

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

cmd/openim-api/main.go

@@ -15,10 +15,9 @@
 package main
 
 import (
-	_ "net/http/pprof"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
 	"github.com/openimsdk/tools/system/program"
+	_ "net/http/pprof"
 )
 
 func main() {

config/discovery.yml

@@ -5,9 +5,4 @@ etcd:
   username: ''
   password: ''
 
-zookeeper:
-  schema: openim
-  address: [ localhost:12181 ]
-  username: ''
-  password: ''
 

config/share.yml

@@ -13,4 +13,17 @@ rpcRegisterName:
 imAdminUserID: [ imAdmin ]
 
 # 1: For Android, iOS, Windows, Mac, and web platforms, only one instance can be online at a time
-multiLoginPolicy: 1
+multiLogin:
+  policy: 1
+  maxNumOneEnd: 30
+  customizeLoginNum:
+    ios: 1
+    android: 1
+    windows: 1
+    osx: 1
+    web: 1
+    miniWeb: 1
+    linux: 1
+    aPad: 1
+    iPad: 1
+    admin: 1

docker-compose.yml

@@ -43,19 +43,6 @@ services:
     networks:
       - openim
 
-  zookeeper:
-    image: "${ZOOKEEPER_IMAGE}"
-    container_name: zookeeper
-    ports:
-      - "12181:2181"
-    environment:
-      #JVMFLAGS: "-Xms32m -Xmx128m"
-      TZ: "Asia/Shanghai"
-      ALLOW_ANONYMOUS_LOGIN: "yes"
-    restart: always
-    networks:
-      - openim
-
   etcd:
     image: "${ETCD_IMAGE}"
     container_name: etcd
@@ -142,6 +129,7 @@ services:
 #    image: ${PROMETHEUS_IMAGE}
 #    container_name: prometheus
 #    restart: always
+#    user: root
 #    volumes:
 #      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
 #      - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml

go.mod

@@ -12,7 +12,7 @@ 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
+	github.com/openimsdk/protocol v0.0.72-alpha.46
 	github.com/openimsdk/tools v0.0.50-alpha.16
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0

go.sum

@@ -319,8 +319,8 @@ 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.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.72 h1:K+vslwaR7lDXyBzb07UuEQITaqsgighz7NyXVIWsu6A=
+github.com/openimsdk/protocol v0.0.72-alpha.46 h1:1LZlfEHLzw1F4afFmqBczmXKJWm5rUQ+yr8rJ4oyEAc=
-github.com/openimsdk/protocol v0.0.72/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+github.com/openimsdk/protocol v0.0.72-alpha.46/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
 github.com/openimsdk/tools v0.0.50-alpha.16 h1:bC1AQvJMuOHtZm8LZRvN8L5mH1Ws2VYdL+TLTs1iGSc=
 github.com/openimsdk/tools v0.0.50-alpha.16/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=

internal/api/jssdk/jssdk.go

@@ -1,11 +1,15 @@
 package jssdk
 
 import (
+	"context"
 	"github.com/gin-gonic/gin"
 	"github.com/openimsdk/protocol/conversation"
+	"github.com/openimsdk/protocol/group"
+	"github.com/openimsdk/protocol/jssdk"
 	"github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/protocol/sdkws"
-	"github.com/openimsdk/tools/a2r"
+	"github.com/openimsdk/protocol/user"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
 	"sort"
@@ -16,16 +20,22 @@ const (
 	defaultGetActiveConversation = 100
 )
 
-func NewJSSdkApi(msg msg.MsgClient, conv conversation.ConversationClient) *JSSdk {
+func NewJSSdkApi(user user.UserClient, friend relation.FriendClient, group group.GroupClient, msg msg.MsgClient, conv conversation.ConversationClient) *JSSdk {
 	return &JSSdk{
-		msg:  msg,
+		user:   user,
-		conv: conv,
+		friend: friend,
+		group:  group,
+		msg:    msg,
+		conv:   conv,
 	}
 }
 
 type JSSdk struct {
-	msg  msg.MsgClient
+	user   user.UserClient
-	conv conversation.ConversationClient
+	friend relation.FriendClient
+	group  group.GroupClient
+	msg    msg.MsgClient
+	conv   conversation.ConversationClient
 }
 
 func (x *JSSdk) GetActiveConversations(c *gin.Context) {
@@ -36,25 +46,71 @@ func (x *JSSdk) GetConversations(c *gin.Context) {
 	call(c, x.getConversations)
 }
 
-func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, error) {
+func (x *JSSdk) fillConversations(ctx context.Context, conversations []*jssdk.ConversationMsg) error {
-	req, err := a2r.ParseRequest[ActiveConversationsReq](ctx)
+	if len(conversations) == 0 {
-	if err != nil {
+		return nil
-		return nil, err
 	}
+	var (
+		userIDs  []string
+		groupIDs []string
+	)
+	for _, c := range conversations {
+		if c.Conversation.GroupID == "" {
+			userIDs = append(userIDs, c.Conversation.UserID)
+		} else {
+			groupIDs = append(groupIDs, c.Conversation.GroupID)
+		}
+	}
+	var (
+		userMap   map[string]*sdkws.UserInfo
+		friendMap map[string]*relation.FriendInfoOnly
+		groupMap  map[string]*sdkws.GroupInfo
+	)
+	if len(userIDs) > 0 {
+		users, err := field(ctx, x.user.GetDesignateUsers, &user.GetDesignateUsersReq{UserIDs: userIDs}, (*user.GetDesignateUsersResp).GetUsersInfo)
+		if err != nil {
+			return err
+		}
+		friends, err := field(ctx, x.friend.GetFriendInfo, &relation.GetFriendInfoReq{OwnerUserID: conversations[0].Conversation.OwnerUserID, FriendUserIDs: userIDs}, (*relation.GetFriendInfoResp).GetFriendInfos)
+		if err != nil {
+			return err
+		}
+		userMap = datautil.SliceToMap(users, (*sdkws.UserInfo).GetUserID)
+		friendMap = datautil.SliceToMap(friends, (*relation.FriendInfoOnly).GetFriendUserID)
+	}
+	if len(groupIDs) > 0 {
+		resp, err := x.group.GetGroupsInfo(ctx, &group.GetGroupsInfoReq{GroupIDs: groupIDs})
+		if err != nil {
+			return err
+		}
+		groupMap = datautil.SliceToMap(resp.GroupInfos, (*sdkws.GroupInfo).GetGroupID)
+	}
+	for _, c := range conversations {
+		if c.Conversation.GroupID == "" {
+			c.User = userMap[c.Conversation.UserID]
+			c.Friend = friendMap[c.Conversation.UserID]
+		} else {
+			c.Group = groupMap[c.Conversation.GroupID]
+		}
+	}
+	return nil
+}
+
+func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActiveConversationsReq) (*jssdk.GetActiveConversationsResp, error) {
 	if req.Count <= 0 || req.Count > maxGetActiveConversation {
 		req.Count = defaultGetActiveConversation
 	}
-	opUserID := mcontext.GetOpUserID(ctx)
+	req.OwnerUserID = mcontext.GetOpUserID(ctx)
 	conversationIDs, err := field(ctx, x.conv.GetConversationIDs,
-		&conversation.GetConversationIDsReq{UserID: opUserID}, (*conversation.GetConversationIDsResp).GetConversationIDs)
+		&conversation.GetConversationIDsReq{UserID: req.OwnerUserID}, (*conversation.GetConversationIDsResp).GetConversationIDs)
 	if err != nil {
 		return nil, err
 	}
 	if len(conversationIDs) == 0 {
-		return &ConversationsResp{}, nil
+		return &jssdk.GetActiveConversationsResp{}, nil
 	}
 	readSeq, err := field(ctx, x.msg.GetHasReadSeqs,
-		&msg.GetHasReadSeqsReq{UserID: opUserID, ConversationIDs: conversationIDs}, (*msg.SeqsInfoResp).GetMaxSeqs)
+		&msg.GetHasReadSeqsReq{UserID: req.OwnerUserID, ConversationIDs: conversationIDs}, (*msg.SeqsInfoResp).GetMaxSeqs)
 	if err != nil {
 		return nil, err
 	}
@@ -64,24 +120,24 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er
 		return nil, err
 	}
 	if len(activeConversation) == 0 {
-		return &ConversationsResp{}, nil
+		return &jssdk.GetActiveConversationsResp{}, nil
 	}
 	sortConversations := sortActiveConversations{
 		Conversation: activeConversation,
 	}
 	if len(activeConversation) > 1 {
 		pinnedConversationIDs, err := field(ctx, x.conv.GetPinnedConversationIDs,
-			&conversation.GetPinnedConversationIDsReq{UserID: opUserID}, (*conversation.GetPinnedConversationIDsResp).GetConversationIDs)
+			&conversation.GetPinnedConversationIDsReq{UserID: req.OwnerUserID}, (*conversation.GetPinnedConversationIDsResp).GetConversationIDs)
 		if err != nil {
 			return nil, err
 		}
 		sortConversations.PinnedConversationIDs = datautil.SliceSet(pinnedConversationIDs)
 	}
 	sort.Sort(&sortConversations)
-	sortList := sortConversations.Top(req.Count)
+	sortList := sortConversations.Top(int(req.Count))
 	conversations, err := field(ctx, x.conv.GetConversations,
 		&conversation.GetConversationsReq{
-			OwnerUserID: opUserID,
+			OwnerUserID: req.OwnerUserID,
 			ConversationIDs: datautil.Slice(sortList, func(c *msg.ActiveConversation) string {
 				return c.ConversationID
 			})}, (*conversation.GetConversationsResp).GetConversations)
@@ -90,7 +146,7 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er
 	}
 	msgs, err := field(ctx, x.msg.GetSeqMessage,
 		&msg.GetSeqMessageReq{
-			UserID: opUserID,
+			UserID: req.OwnerUserID,
 			Conversations: datautil.Slice(sortList, func(c *msg.ActiveConversation) *msg.ConversationSeqs {
 				return &msg.ConversationSeqs{
 					ConversationID: c.ConversationID,
@@ -104,7 +160,7 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er
 	conversationMap := datautil.SliceToMap(conversations, func(c *conversation.Conversation) string {
 		return c.ConversationID
 	})
-	resp := make([]ConversationMsg, 0, len(sortList))
+	resp := make([]*jssdk.ConversationMsg, 0, len(sortList))
 	for _, c := range sortList {
 		conv, ok := conversationMap[c.ConversationID]
 		if !ok {
@@ -114,13 +170,16 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er
 		if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 {
 			lastMsg = msgList.Msgs[0]
 		}
-		resp = append(resp, ConversationMsg{
+		resp = append(resp, &jssdk.ConversationMsg{
 			Conversation: conv,
 			LastMsg:      lastMsg,
 			MaxSeq:       c.MaxSeq,
 			ReadSeq:      readSeq[c.ConversationID],
 		})
 	}
+	if err := x.fillConversations(ctx, resp); err != nil {
+		return nil, err
+	}
 	var unreadCount int64
 	for _, c := range activeConversation {
 		count := c.MaxSeq - readSeq[c.ConversationID]
@@ -128,24 +187,20 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er
 			unreadCount += count
 		}
 	}
-	return &ConversationsResp{
+	return &jssdk.GetActiveConversationsResp{
 		Conversations: resp,
 		UnreadCount:   unreadCount,
 	}, nil
 }
 
-func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) {
+func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversationsReq) (*jssdk.GetConversationsResp, error) {
-	req, err := a2r.ParseRequest[conversation.GetConversationsReq](ctx)
-	if err != nil {
-		return nil, err
-	}
 	req.OwnerUserID = mcontext.GetOpUserID(ctx)
-	conversations, err := field(ctx, x.conv.GetConversations, req, (*conversation.GetConversationsResp).GetConversations)
+	conversations, err := field(ctx, x.conv.GetConversations, &conversation.GetConversationsReq{OwnerUserID: req.OwnerUserID, ConversationIDs: req.ConversationIDs}, (*conversation.GetConversationsResp).GetConversations)
 	if err != nil {
 		return nil, err
 	}
 	if len(conversations) == 0 {
-		return &ConversationsResp{}, nil
+		return &jssdk.GetConversationsResp{}, nil
 	}
 	req.ConversationIDs = datautil.Slice(conversations, func(c *conversation.Conversation) string {
 		return c.ConversationID
@@ -177,19 +232,22 @@ func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) {
 			return nil, err
 		}
 	}
-	resp := make([]ConversationMsg, 0, len(conversations))
+	resp := make([]*jssdk.ConversationMsg, 0, len(conversations))
 	for _, c := range conversations {
 		var lastMsg *sdkws.MsgData
 		if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 {
 			lastMsg = msgList.Msgs[0]
 		}
-		resp = append(resp, ConversationMsg{
+		resp = append(resp, &jssdk.ConversationMsg{
 			Conversation: c,
 			LastMsg:      lastMsg,
 			MaxSeq:       maxSeqs[c.ConversationID],
 			ReadSeq:      readSeqs[c.ConversationID],
 		})
 	}
+	if err := x.fillConversations(ctx, resp); err != nil {
+		return nil, err
+	}
 	var unreadCount int64
 	for conversationID, maxSeq := range maxSeqs {
 		count := maxSeq - readSeqs[conversationID]
@@ -197,7 +255,7 @@ func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) {
 			unreadCount += count
 		}
 	}
-	return &ConversationsResp{
+	return &jssdk.GetConversationsResp{
 		Conversations: resp,
 		UnreadCount:   unreadCount,
 	}, nil

internal/api/jssdk/stu.go

@@ -1,22 +0,0 @@
-package jssdk
-
-import (
-	"github.com/openimsdk/protocol/conversation"
-	"github.com/openimsdk/protocol/sdkws"
-)
-
-type ActiveConversationsReq struct {
-	Count int `json:"count"`
-}
-
-type ConversationMsg struct {
-	Conversation *conversation.Conversation `json:"conversation"`
-	LastMsg      *sdkws.MsgData             `json:"lastMsg"`
-	MaxSeq       int64                      `json:"maxSeq"`
-	ReadSeq      int64                      `json:"readSeq"`
-}
-
-type ConversationsResp struct {
-	UnreadCount   int64             `json:"unreadCount"`
-	Conversations []ConversationMsg `json:"conversations"`
-}

internal/api/jssdk/tools.go

@@ -3,8 +3,14 @@ package jssdk
 import (
 	"context"
 	"github.com/gin-gonic/gin"
+	"github.com/openimsdk/tools/a2r"
 	"github.com/openimsdk/tools/apiresp"
+	"github.com/openimsdk/tools/checker"
+	"github.com/openimsdk/tools/errs"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+	"io"
+	"strings"
 )
 
 func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A, opts ...grpc.CallOption) (*B, error), req *A, get func(*B) C) (C, error) {
@@ -16,11 +22,56 @@ func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A
 	return get(resp), nil
 }
 
-func call[R any](c *gin.Context, fn func(ctx *gin.Context) (R, error)) {
+func call[A, B any](c *gin.Context, fn func(ctx context.Context, req *A) (*B, error)) {
-	resp, err := fn(c)
+	var isJSON bool
+	switch contentType := c.GetHeader("Content-Type"); {
+	case contentType == "":
+		isJSON = true
+	case strings.Contains(contentType, "application/json"):
+		isJSON = true
+	case strings.Contains(contentType, "application/protobuf"):
+	case strings.Contains(contentType, "application/x-protobuf"):
+	default:
+		apiresp.GinError(c, errs.ErrArgs.WrapMsg("unsupported content type"))
+		return
+	}
+	var req *A
+	if isJSON {
+		var err error
+		req, err = a2r.ParseRequest[A](c)
+		if err != nil {
+			apiresp.GinError(c, err)
+			return
+		}
+	} else {
+		body, err := io.ReadAll(c.Request.Body)
+		if err != nil {
+			apiresp.GinError(c, err)
+			return
+		}
+		req = new(A)
+		if err := proto.Unmarshal(body, any(req).(proto.Message)); err != nil {
+			apiresp.GinError(c, err)
+			return
+		}
+		if err := checker.Validate(&req); err != nil {
+			apiresp.GinError(c, err)
+			return
+		}
+	}
+	resp, err := fn(c, req)
 	if err != nil {
 		apiresp.GinError(c, err)
 		return
 	}
-	apiresp.GinSuccess(c, resp)
+	if isJSON {
+		apiresp.GinSuccess(c, resp)
+		return
+	}
+	body, err := proto.Marshal(any(resp).(proto.Message))
+	if err != nil {
+		apiresp.GinError(c, err)
+		return
+	}
+	apiresp.GinSuccess(c, body)
 }

internal/api/jssdk_test.go

@@ -1,37 +0,0 @@
-package api
-
-import (
-	"github.com/openimsdk/protocol/msg"
-	"sort"
-	"testing"
-)
-
-func TestName(t *testing.T) {
-	val := sortActiveConversations{
-		Conversation: []*msg.ActiveConversation{
-			{
-				ConversationID: "100",
-				LastTime:       100,
-			},
-			{
-				ConversationID: "200",
-				LastTime:       200,
-			},
-			{
-				ConversationID: "300",
-				LastTime:       300,
-			},
-			{
-				ConversationID: "400",
-				LastTime:       400,
-			},
-		},
-		//PinnedConversationIDs: map[string]struct{}{
-		//	"100": {},
-		//	"300": {},
-		//},
-	}
-	sort.Sort(&val)
-	t.Log(val)
-
-}

internal/api/router.go

@@ -77,7 +77,7 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
 	r.Use(prommetricsGin(), gin.Recovery(), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc))
 	u := NewUserApi(*userRpc)
 	m := NewMessageApi(messageRpc, userRpc, config.Share.IMAdminUserID)
-	j := jssdk.NewJSSdkApi(messageRpc.Client, conversationRpc.Client)
+	j := jssdk.NewJSSdkApi(userRpc.Client, friendRpc.Client, groupRpc.Client, messageRpc.Client, conversationRpc.Client)
 	userRouterGroup := r.Group("/user")
 	{
 		userRouterGroup.POST("/user_register", u.UserRegister)

internal/msggateway/encoder.go

@@ -15,8 +15,7 @@
 package msggateway
 
 import (
-	"bytes"
+	"encoding/json"
-	"encoding/gob"
 
 	"github.com/openimsdk/tools/errs"
 )
@@ -33,19 +32,17 @@ func NewGobEncoder() *GobEncoder {
 }
 
 func (g *GobEncoder) Encode(data any) ([]byte, error) {
-	buff := bytes.Buffer{}
+	b, err := json.Marshal(data)
-	enc := gob.NewEncoder(&buff)
+	if err != nil {
-	if err := enc.Encode(data); err != nil {
+		return nil, errs.New("Encoder.Encode failed", "action", "encode")
-		return nil, errs.WrapMsg(err, "GobEncoder.Encode failed", "action", "encode")
 	}
-	return buff.Bytes(), nil
+	return b, nil
 }
 
 func (g *GobEncoder) Decode(encodeData []byte, decodeData any) error {
-	buff := bytes.NewBuffer(encodeData)
+	err := json.Unmarshal(encodeData, decodeData)
-	dec := gob.NewDecoder(buff)
+	if err != nil {
-	if err := dec.Decode(decodeData); err != nil {
+		return errs.New("Encoder.Decode failed", "action", "decode")
-		return errs.WrapMsg(err, "GobEncoder.Decode failed", "action", "decode")
 	}
 	return nil
 }

internal/msggateway/encoder_test.go

@@ -0,0 +1,48 @@
+package msggateway
+
+import (
+	"testing"
+)
+
+func TestGobEncoder_Encode(t *testing.T) {
+	encoder := NewGobEncoder()
+
+	// 测试用例1: 编码 []byte 数据
+	inputData := []byte("test data")
+	encodedData, err := encoder.Encode(inputData)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if string(encodedData) != string(inputData) {
+		t.Fatalf("expected encoded data to be '%s', got '%s'", inputData, encodedData)
+	}
+
+	// 测试用例2: 编码非 []byte 数据
+	nonByteData := "string data"
+	_, err = encoder.Encode(nonByteData)
+	if err == nil {
+		t.Fatalf("expected an error when encoding non-byte data, got none")
+	}
+}
+
+func TestGobEncoder_Decode(t *testing.T) {
+	encoder := NewGobEncoder()
+
+	// 测试用例1: 解码到 []byte 数据
+	encodedData := []byte("test data")
+	var decodedData []byte
+	err := encoder.Decode(encodedData, &decodedData)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if string(decodedData) != string(encodedData) {
+		t.Fatalf("expected decoded data to be '%s', got '%s'", encodedData, decodedData)
+	}
+
+	// 测试用例2: 解码到非 []byte 数据
+	var nonByteData string
+	err = encoder.Decode(encodedData, &nonByteData)
+	if err == nil {
+		t.Fatalf("expected an error when decoding to non-byte data, got none")
+	}
+}

internal/msggateway/hub_server.go

@@ -155,6 +155,7 @@ func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.M
 			(client.IsBackground && client.PlatformID != constant.IOSPlatformID) {
 			err := client.PushMessage(ctx, msgData)
 			if err != nil {
+				log.ZWarn(ctx, "online push msg failed", err, "userID", userID, "platformID", client.PlatformID)
 				userPlatform.ResultCode = int64(servererrs.ErrPushMsgErr.Code())
 			} else {
 				if _, ok := s.pushTerminal[client.PlatformID]; ok {

internal/msggateway/online.go

@@ -90,6 +90,19 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
 		if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil {
 			log.ZError(ctx, "update user online status", err)
 		}
+		for _, ss := range req.Status {
+			for _, online := range ss.Online {
+				client, _, _ := ws.clients.Get(ss.UserID, int(online))
+				back := false
+				if len(client) > 0 {
+					back = client[0].IsBackground
+				}
+				ws.webhookAfterUserOnline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOnline, ss.UserID, int(online), back, ss.ConnID)
+			}
+			for _, offline := range ss.Offline {
+				ws.webhookAfterUserOffline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOffline, ss.UserID, int(offline), ss.ConnID)
+			}
+		}
 	}
 
 	for i := 0; i < concurrent; i++ {

internal/msggateway/ws_server.go

@@ -1,17 +1,3 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package msggateway
 
 import (
@@ -212,7 +198,6 @@ func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *C
 	if err != nil {
 		return err
 	}
-
 	wg := errgroup.Group{}
 	wg.SetLimit(concurrentRequest)
 
@@ -321,8 +306,32 @@ func (ws *WsServer) KickUserConn(client *Client) error {
 }
 
 func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Client, newClient *Client) {
-	switch ws.msgGatewayConfig.Share.MultiLoginPolicy {
+	kickTokenFunc := func(kickClients []*Client) {
+		var kickTokens []string
+		ws.clients.DeleteClients(newClient.UserID, kickClients)
+		for _, c := range kickClients {
+			kickTokens = append(kickTokens, c.token)
+			err := c.KickOnlineMessage()
+			if err != nil {
+				log.ZWarn(c.ctx, "KickOnlineMessage", err)
+			}
+		}
+		ctx := mcontext.WithMustInfoCtx(
+			[]string{newClient.ctx.GetOperationID(), newClient.ctx.GetUserID(),
+				constant.PlatformIDToName(newClient.PlatformID), newClient.ctx.GetConnID()},
+		)
+		if _, err := ws.authClient.KickTokens(ctx, kickTokens); err != nil {
+			log.ZWarn(newClient.ctx, "kickTokens err", err)
+		}
+	}
+
+	switch ws.msgGatewayConfig.Share.MultiLogin.Policy {
 	case constant.DefalutNotKick:
+	case constant.WebAndOther:
+		if constant.PlatformIDToClass(newClient.PlatformID) == constant.WebPlatformStr {
+			return
+		}
+		fallthrough
 	case constant.PCAndOther:
 		if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC {
 			return
@@ -347,6 +356,35 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 			log.ZWarn(newClient.ctx, "InvalidateToken err", err, "userID", newClient.UserID,
 				"platformID", newClient.PlatformID)
 		}
+	case constant.PcMobileAndWeb:
+		clients, ok := ws.clients.GetAll(newClient.UserID)
+		if !ok {
+			return
+		}
+		var (
+			kickClients []*Client
+		)
+		for _, client := range clients {
+			if constant.PlatformIDToClass(client.PlatformID) == constant.PlatformIDToClass(newClient.PlatformID) {
+				kickClients = append(kickClients, client)
+			}
+		}
+		kickTokenFunc(kickClients)
+
+	case constant.SingleTerminalLogin:
+		clients, ok := ws.clients.GetAll(newClient.UserID)
+		if !ok {
+			return
+		}
+		var (
+			kickClients []*Client
+		)
+		for _, client := range clients {
+			kickClients = append(kickClients, client)
+		}
+		kickTokenFunc(kickClients)
+	case constant.Customize:
+		// todo
 	}
 }
 

internal/push/a_test.go

@@ -1,29 +0,0 @@
-package push
-
-import (
-	"github.com/openimsdk/protocol/sdkws"
-	"testing"
-)
-
-func TestName(t *testing.T) {
-	var c ConsumerHandler
-	c.readCh = make(chan *sdkws.MarkAsReadTips)
-
-	go c.loopRead()
-
-	go func() {
-		for i := 0; ; i++ {
-			seq := int64(i + 1)
-			if seq%3 == 0 {
-				seq = 1
-			}
-			c.readCh <- &sdkws.MarkAsReadTips{
-				ConversationID:   "c100",
-				MarkAsReadUserID: "u100",
-				HasReadSeq:       seq,
-			}
-		}
-	}()
-
-	select {}
-}

internal/push/offlinepush/offlinepusher.go

@@ -23,10 +23,13 @@ import (
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mcontext"
+	"strings"
 )
 
 const (
-	geTUI    = "geTui"
+	geTUI    = "getui"
 	firebase = "fcm"
 	jPush    = "jpush"
 )
@@ -38,6 +41,7 @@ type OfflinePusher interface {
 
 func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (OfflinePusher, error) {
 	var offlinePusher OfflinePusher
+	pushConf.Enable = strings.ToLower(pushConf.Enable)
 	switch pushConf.Enable {
 	case geTUI:
 		offlinePusher = getui.NewClient(pushConf, cache)
@@ -47,6 +51,7 @@ func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPa
 		offlinePusher = jpush.NewClient(pushConf)
 	default:
 		offlinePusher = dummy.NewClient()
+		log.ZWarn(mcontext.WithMustInfoCtx([]string{"push start", "admin", "admin", ""}), "Unknown push config", nil)
 	}
 	return offlinePusher, nil
 }

internal/push/offlinepush_handler.go

@@ -55,6 +55,9 @@ func (o *OfflinePushConsumerHandler) handleMsg2OfflinePush(ctx context.Context,
 		log.ZError(ctx, "offline push msg is empty", errs.New("offlinePushMsg is empty"), "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
 		return
 	}
+	if offlinePushMsg.MsgData.Status == constant.MsgStatusSending {
+		offlinePushMsg.MsgData.Status = constant.MsgStatusSendSuccess
+	}
 	log.ZInfo(ctx, "receive to OfflinePush MQ", "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
 
 	err := o.offlinePushMsg(ctx, offlinePushMsg.MsgData, offlinePushMsg.UserIDs)

internal/push/push_handler.go

@@ -194,6 +194,9 @@ func (c *ConsumerHandler) shouldPushOffline(_ context.Context, msg *sdkws.MsgDat
 }
 
 func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) ([]*msggateway.SingleMsgToUserResults, error) {
+	if msg != nil && msg.Status == constant.MsgStatusSending {
+		msg.Status = constant.MsgStatusSendSuccess
+	}
 	onlineUserIDs, offlineUserIDs, err := c.onlineCache.GetUsersOnline(ctx, pushToUserIDs)
 	if err != nil {
 		return nil, err

internal/rpc/auth/auth.go

@@ -65,7 +65,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 			redis2.NewTokenCacheModel(rdb, config.RpcConfig.TokenPolicy.Expire),
 			config.Share.Secret,
 			config.RpcConfig.TokenPolicy.Expire,
-			config.Share.MultiLoginPolicy,
+			config.Share.MultiLogin,
 		),
 		config: config,
 	})
@@ -230,3 +230,10 @@ func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.Invalidate
 	}
 	return &pbauth.InvalidateTokenResp{}, nil
 }
+
+func (s *authServer) KickTokens(ctx context.Context, req *pbauth.KickTokensReq) (*pbauth.KickTokensResp, error) {
+	if err := s.authDatabase.BatchSetTokenMapByUidPid(ctx, req.Tokens); err != nil {
+		return nil, err
+	}
+	return &pbauth.KickTokensResp{}, nil
+}

internal/rpc/group/group.go

@@ -465,7 +465,7 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 		return nil, err
 	}
 
-	if err = g.notification.MemberEnterNotification(ctx, req.GroupID, req.InvitedUserIDs...); err != nil {
+	if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, opUserID, req.InvitedUserIDs...); err != nil {
 		return nil, err
 	}
 	return &pbgroup.InviteUserToGroupResp{}, nil
@@ -1485,9 +1485,6 @@ func (g *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbgroup.SetGr
 		return nil, errs.ErrNoPermission.WrapMsg("no op user id")
 	}
 	isAppManagerUid := authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID)
-	for i := range req.Members {
-		req.Members[i].FaceURL = nil
-	}
 	groupMembers := make(map[string][]*pbgroup.SetGroupMemberInfo)
 	for i, member := range req.Members {
 		if member.RoleLevel != nil {
@@ -1529,29 +1526,61 @@ func (g *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbgroup.SetGr
 		case 0:
 			if !isAppManagerUid {
 				roleLevel := dbMembers[opUserIndex].RoleLevel
-				if roleLevel != constant.GroupOwner {
+				var (
-					switch roleLevel {
+					dbSelf  = &model.GroupMember{}
-					case constant.GroupAdmin:
+					reqSelf *pbgroup.SetGroupMemberInfo
-						for _, member := range dbMembers {
+				)
-							if member.RoleLevel == constant.GroupOwner {
+				switch roleLevel {
-								return nil, errs.ErrNoPermission.WrapMsg("admin can not change group owner")
+				case constant.GroupOwner:
-							}
+					for _, member := range dbMembers {
-							if member.RoleLevel == constant.GroupAdmin && member.UserID != opUserID {
+						if member.UserID == opUserID {
-								return nil, errs.ErrNoPermission.WrapMsg("admin can not change other group admin")
+							dbSelf = member
-							}
+							break
 						}
-					case constant.GroupOrdinaryUsers:
+					}
-						for _, member := range dbMembers {
+				case constant.GroupAdmin:
-							if !(member.RoleLevel == constant.GroupOrdinaryUsers && member.UserID == opUserID) {
+					for _, member := range dbMembers {
-								return nil, errs.ErrNoPermission.WrapMsg("ordinary users can not change other role level")
+						if member.UserID == opUserID {
-							}
+							dbSelf = member
 						}
-					default:
+						if member.RoleLevel == constant.GroupOwner {
-						for _, member := range dbMembers {
+							return nil, errs.ErrNoPermission.WrapMsg("admin can not change group owner")
-							if member.RoleLevel >= roleLevel {
-								return nil, errs.ErrNoPermission.WrapMsg("can not change higher role level")
-							}
 						}
+						if member.RoleLevel == constant.GroupAdmin && member.UserID != opUserID {
+							return nil, errs.ErrNoPermission.WrapMsg("admin can not change other group admin")
+						}
+					}
+				case constant.GroupOrdinaryUsers:
+					for _, member := range dbMembers {
+						if member.UserID == opUserID {
+							dbSelf = member
+						}
+						if !(member.RoleLevel == constant.GroupOrdinaryUsers && member.UserID == opUserID) {
+							return nil, errs.ErrNoPermission.WrapMsg("ordinary users can not change other role level")
+						}
+					}
+				default:
+					for _, member := range dbMembers {
+						if member.UserID == opUserID {
+							dbSelf = member
+						}
+						if member.RoleLevel >= roleLevel {
+							return nil, errs.ErrNoPermission.WrapMsg("can not change higher role level")
+						}
+					}
+				}
+				for _, member := range req.Members {
+					if member.UserID == opUserID {
+						reqSelf = member
+						break
+					}
+				}
+				if reqSelf != nil && reqSelf.RoleLevel != nil {
+					if reqSelf.RoleLevel.GetValue() > dbSelf.RoleLevel {
+						return nil, errs.ErrNoPermission.WrapMsg("can not improve role level by self")
+					}
+					if roleLevel == constant.GroupOwner {
+						return nil, errs.ErrArgs.WrapMsg("group owner can not change own role level") // Prevent the absence of a group owner
 					}
 				}
 			}
@@ -1592,7 +1621,7 @@ func (g *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbgroup.SetGr
 				g.notification.GroupMemberSetToOrdinaryUserNotification(ctx, member.GroupID, member.UserID)
 			}
 		}
-		if member.Nickname != nil || member.FaceURL != nil || member.Ex != nil {
+		if member.Nickname != nil || member.FaceURL != nil || member.Ex != nil || member.RoleLevel != nil {
 			g.notification.GroupMemberInfoSetNotification(ctx, member.GroupID, member.UserID)
 		}
 	}

internal/rpc/group/notification.go

@@ -38,6 +38,7 @@ import (
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/stringutil"
 	"go.mongodb.org/mongo-driver/mongo"
+	"time"
 )
 
 // GroupApplicationReceiver
@@ -572,8 +573,51 @@ func (g *GroupNotificationSender) GroupApplicationAgreeMemberEnterNotification(c
 	return nil
 }
 
-func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, groupID string, entrantUserID ...string) error {
+func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, groupID string, entrantUserID string) error {
-	return g.GroupApplicationAgreeMemberEnterNotification(ctx, groupID, "", entrantUserID...)
+	var err error
+	defer func() {
+		if err != nil {
+			log.ZError(ctx, stringutil.GetFuncName(1)+" failed", err)
+		}
+	}()
+
+	if !g.config.RpcConfig.EnableHistoryForNewMembers {
+		conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID)
+		maxSeq, err := g.msgRpcClient.GetConversationMaxSeq(ctx, conversationID)
+		if err != nil {
+			return err
+		}
+		if _, err = g.msgRpcClient.SetUserConversationsMinSeq(ctx, &msg.SetUserConversationsMinSeqReq{
+			UserIDs:        []string{entrantUserID},
+			ConversationID: conversationID,
+			Seq:            maxSeq,
+		}); err != nil {
+			return err
+		}
+	}
+
+	if err := g.conversationRpcClient.GroupChatFirstCreateConversation(ctx, groupID, []string{entrantUserID}); err != nil {
+		return err
+	}
+
+	var group *sdkws.GroupInfo
+	group, err = g.getGroupInfo(ctx, groupID)
+	if err != nil {
+		return err
+	}
+	user, err := g.getGroupMember(ctx, groupID, entrantUserID)
+	if err != nil {
+		return err
+	}
+
+	tips := &sdkws.MemberEnterTips{
+		Group:         group,
+		EntrantUser:   user,
+		OperationTime: time.Now().UnixMilli(),
+	}
+	g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
+	g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberEnterNotification, tips)
+	return nil
 }
 
 func (g *GroupNotificationSender) GroupDismissedNotification(ctx context.Context, tips *sdkws.GroupDismissedTips) {

internal/rpc/relation/friend.go

@@ -273,7 +273,14 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *relation.SetFri
 	return &relation.SetFriendRemarkResp{}, nil
 }
 
-// ok.
+func (s *friendServer) GetFriendInfo(ctx context.Context, req *relation.GetFriendInfoReq) (*relation.GetFriendInfoResp, error) {
+	friends, err := s.db.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs)
+	if err != nil {
+		return nil, err
+	}
+	return &relation.GetFriendInfoResp{FriendInfos: convert.FriendOnlyDB2PbOnly(friends)}, nil
+}
+
 func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *relation.GetDesignatedFriendsReq) (resp *relation.GetDesignatedFriendsResp, err error) {
 	resp = &relation.GetDesignatedFriendsResp{}
 	if datautil.Duplicate(req.FriendUserIDs) {

pkg/common/config/config.go

@@ -361,11 +361,29 @@ type AfterConfig struct {
 }
 
 type Share struct {
-	Secret           string          `mapstructure:"secret"`
+	Secret          string          `mapstructure:"secret"`
-	RpcRegisterName  RpcRegisterName `mapstructure:"rpcRegisterName"`
+	RpcRegisterName RpcRegisterName `mapstructure:"rpcRegisterName"`
-	IMAdminUserID    []string        `mapstructure:"imAdminUserID"`
+	IMAdminUserID   []string        `mapstructure:"imAdminUserID"`
-	MultiLoginPolicy int             `mapstructure:"multiLoginPolicy"`
+	MultiLogin      MultiLogin      `mapstructure:"multiLogin"`
 }
+
+type MultiLogin struct {
+	Policy            int `mapstructure:"policy"`
+	MaxNumOneEnd      int `mapstructure:"maxNumOneEnd"`
+	CustomizeLoginNum struct {
+		IOS     int `mapstructure:"ios"`
+		Android int `mapstructure:"android"`
+		Windows int `mapstructure:"windows"`
+		OSX     int `mapstructure:"osx"`
+		Web     int `mapstructure:"web"`
+		MiniWeb int `mapstructure:"miniWeb"`
+		Linux   int `mapstructure:"linux"`
+		APad    int `mapstructure:"aPad"`
+		IPad    int `mapstructure:"iPad"`
+		Admin   int `mapstructure:"admin"`
+	} `mapstructure:"customizeLoginNum"`
+}
+
 type RpcRegisterName struct {
 	User           string `mapstructure:"user"`
 	Friend         string `mapstructure:"friend"`

pkg/common/convert/friend.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/protocol/relation"
 
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/utils/datautil"
@@ -35,9 +36,7 @@ func FriendPb2DB(friend *sdkws.FriendInfo) *model.Friend {
 	return dbFriend
 }
 
-func FriendDB2Pb(ctx context.Context, friendDB *model.Friend,
+func FriendDB2Pb(ctx context.Context, friendDB *model.Friend, getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error)) (*sdkws.FriendInfo, error) {
-	getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error),
-) (*sdkws.FriendInfo, error) {
 	users, err := getUsers(ctx, []string{friendDB.FriendUserID})
 	if err != nil {
 		return nil, err
@@ -53,11 +52,7 @@ func FriendDB2Pb(ctx context.Context, friendDB *model.Friend,
 	}, nil
 }
 
-func FriendsDB2Pb(
+func FriendsDB2Pb(ctx context.Context, friendsDB []*model.Friend, getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error)) (friendsPb []*sdkws.FriendInfo, err error) {
-	ctx context.Context,
-	friendsDB []*model.Friend,
-	getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error),
-) (friendsPb []*sdkws.FriendInfo, err error) {
 	if len(friendsDB) == 0 {
 		return nil, nil
 	}
@@ -86,7 +81,21 @@ func FriendsDB2Pb(
 		friendsPb = append(friendsPb, friendPb)
 	}
 	return friendsPb, nil
+}
 
+func FriendOnlyDB2PbOnly(friendsDB []*model.Friend) []*relation.FriendInfoOnly {
+	return datautil.Slice(friendsDB, func(f *model.Friend) *relation.FriendInfoOnly {
+		return &relation.FriendInfoOnly{
+			OwnerUserID:    f.OwnerUserID,
+			FriendUserID:   f.FriendUserID,
+			Remark:         f.Remark,
+			CreateTime:     f.CreateTime.UnixMilli(),
+			AddSource:      f.AddSource,
+			OperatorUserID: f.OperatorUserID,
+			Ex:             f.Ex,
+			IsPinned:       f.IsPinned,
+		}
+	})
 }
 
 func FriendRequestDB2Pb(ctx context.Context, friendRequests []*model.FriendRequest, getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error)) ([]*sdkws.FriendRequest, error) {

pkg/common/storage/cache/cachekey/token.go

@@ -1,6 +1,9 @@
 package cachekey
 
-import "github.com/openimsdk/protocol/constant"
+import (
+	"github.com/openimsdk/protocol/constant"
+	"strings"
+)
 
 const (
 	UidPidToken = "UID_PID_TOKEN_STATUS:"
@@ -9,3 +12,17 @@ const (
 func GetTokenKey(userID string, platformID int) string {
 	return UidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
 }
+
+func GetAllPlatformTokenKey(userID string) []string {
+	res := make([]string, len(constant.PlatformID2Name))
+	for k := range constant.PlatformID2Name {
+		res[k-1] = GetTokenKey(userID, k)
+	}
+	return res
+}
+
+func GetPlatformIDByTokenKey(key string) int {
+	splitKey := strings.Split(key, ":")
+	platform := splitKey[len(splitKey)-1]
+	return constant.PlatformNameToID(platform)
+}

pkg/common/storage/cache/redis/token.go

@@ -1,17 +1,3 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package redis
 
 import (
@@ -21,6 +7,7 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/redis/go-redis/v9"
 	"strconv"
+	"sync"
 	"time"
 )
 
@@ -67,6 +54,43 @@ func (c *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
 	return mm, nil
 }
 
+func (c *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
+	var (
+		res     = make(map[int]map[string]int)
+		resLock = sync.Mutex{}
+	)
+
+	keys := cachekey.GetAllPlatformTokenKey(userID)
+	if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
+		pipe := c.rdb.Pipeline()
+		mapRes := make([]*redis.MapStringStringCmd, len(keys))
+		for i, key := range keys {
+			mapRes[i] = pipe.HGetAll(ctx, key)
+		}
+		_, err := pipe.Exec(ctx)
+		if err != nil {
+			return err
+		}
+		for i, m := range mapRes {
+			mm := make(map[string]int)
+			for k, v := range m.Val() {
+				state, err := strconv.Atoi(v)
+				if err != nil {
+					return errs.WrapMsg(err, "redis token value is not int", "value", v, "userID", userID)
+				}
+				mm[k] = state
+			}
+			resLock.Lock()
+			res[cachekey.GetPlatformIDByTokenKey(keys[i])] = mm
+			resLock.Unlock()
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	return res, nil
+}
+
 func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
 	mm := make(map[string]any)
 	for k, v := range m {
@@ -75,6 +99,18 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla
 	return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err())
 }
 
+func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]int) error {
+	pipe := c.rdb.Pipeline()
+	for k, v := range tokens {
+		pipe.HSet(ctx, k, v)
+	}
+	_, err := pipe.Exec(ctx)
+	if err != nil {
+		return errs.Wrap(err)
+	}
+	return nil
+}
+
 func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error {
 	return errs.Wrap(c.rdb.HDel(ctx, cachekey.GetTokenKey(userID, platformID), fields...).Err())
 }

pkg/common/storage/cache/token.go

@@ -9,6 +9,8 @@ type TokenModel interface {
 	// SetTokenFlagEx set token and flag with expire time
 	SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error
 	GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
+	GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error)
 	SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
+	BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]int) error
 	DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
 }

pkg/common/storage/controller/auth.go

@@ -1,21 +1,9 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package controller
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"github.com/openimsdk/tools/log"
 
 	"github.com/golang-jwt/jwt/v4"
@@ -32,18 +20,41 @@ type AuthDatabase interface {
 	// Create token
 	CreateToken(ctx context.Context, userID string, platformID int) (string, error)
 
+	BatchSetTokenMapByUidPid(ctx context.Context, tokens []string) error
+
 	SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
 }
 
-type authDatabase struct {
+type multiLoginConfig struct {
-	cache            cache.TokenModel
+	Policy            int
-	accessSecret     string
+	MaxNumOneEnd      int
-	accessExpire     int64
+	CustomizeLoginNum map[int]int
-	multiLoginPolicy int
 }
 
-func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, policy int) AuthDatabase {
+type authDatabase struct {
-	return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLoginPolicy: policy}
+	cache        cache.TokenModel
+	accessSecret string
+	accessExpire int64
+	multiLogin   multiLoginConfig
+}
+
+func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin) AuthDatabase {
+	return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLogin: multiLoginConfig{
+		Policy:       multiLogin.Policy,
+		MaxNumOneEnd: multiLogin.MaxNumOneEnd,
+		CustomizeLoginNum: map[int]int{
+			constant.IOSPlatformID:        multiLogin.CustomizeLoginNum.IOS,
+			constant.AndroidPlatformID:    multiLogin.CustomizeLoginNum.Android,
+			constant.WindowsPlatformID:    multiLogin.CustomizeLoginNum.Windows,
+			constant.OSXPlatformID:        multiLogin.CustomizeLoginNum.OSX,
+			constant.WebPlatformID:        multiLogin.CustomizeLoginNum.Web,
+			constant.MiniWebPlatformID:    multiLogin.CustomizeLoginNum.MiniWeb,
+			constant.LinuxPlatformID:      multiLogin.CustomizeLoginNum.Linux,
+			constant.AndroidPadPlatformID: multiLogin.CustomizeLoginNum.APad,
+			constant.IPadPlatformID:       multiLogin.CustomizeLoginNum.IPad,
+			constant.AdminPlatformID:      multiLogin.CustomizeLoginNum.Admin,
+		},
+	}}
 }
 
 // If the result is empty.
@@ -55,22 +66,38 @@ func (a *authDatabase) SetTokenMapByUidPid(ctx context.Context, userID string, p
 	return a.cache.SetTokenMapByUidPid(ctx, userID, platformID, m)
 }
 
+func (a *authDatabase) BatchSetTokenMapByUidPid(ctx context.Context, tokens []string) error {
+	setMap := make(map[string]map[string]int)
+	for _, token := range tokens {
+		claims, err := tokenverify.GetClaimFromToken(token, authverify.Secret(a.accessSecret))
+		key := cachekey.GetTokenKey(claims.UserID, claims.PlatformID)
+		if err != nil {
+			continue
+		} else {
+			if v, ok := setMap[key]; ok {
+				v[token] = constant.KickedToken
+			} else {
+				setMap[key] = map[string]int{
+					token: constant.KickedToken,
+				}
+			}
+		}
+	}
+	if err := a.cache.BatchSetTokenMapByUidPid(ctx, setMap); err != nil {
+		return err
+	}
+	return nil
+}
+
 // Create Token.
 func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) {
-	// todo: get all platform token
+	tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID)
-	tokens, err := a.cache.GetTokensWithoutError(ctx, userID, platformID)
 	if err != nil {
 		return "", err
 	}
-	var deleteTokenKey []string
+	deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
-	var kickedTokenKey []string
+	if err != nil {
-	for k, v := range tokens {
+		return "", err
-		t, err := tokenverify.GetClaimFromToken(k, authverify.Secret(a.accessSecret))
-		if err != nil || v != constant.NormalToken {
-			deleteTokenKey = append(deleteTokenKey, k)
-		} else if a.checkKickToken(ctx, platformID, t) {
-			kickedTokenKey = append(kickedTokenKey, k)
-		}
 	}
 	if len(deleteTokenKey) != 0 {
 		err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
@@ -78,16 +105,6 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
 			return "", err
 		}
 	}
-
-	const adminTokenMaxNum = 30
-	if platformID == constant.AdminPlatformID {
-		if len(kickedTokenKey) > adminTokenMaxNum {
-			kickedTokenKey = kickedTokenKey[:len(kickedTokenKey)-adminTokenMaxNum]
-		} else {
-			kickedTokenKey = nil
-		}
-	}
-
 	if len(kickedTokenKey) != 0 {
 		for _, k := range kickedTokenKey {
 			err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
@@ -111,22 +128,140 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
 	return tokenString, nil
 }
 
-func (a *authDatabase) checkKickToken(ctx context.Context, platformID int, token *tokenverify.Claims) bool {
+func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string]int, platformID int) ([]string, []string, error) {
-	switch a.multiLoginPolicy {
+	// todo: Move the logic for handling old data to another location.
-	case constant.DefalutNotKick:
+	var (
-		return false
+		loginTokenMap  = make(map[int][]string) // The length of the value of the map must be greater than 0
-	case constant.PCAndOther:
+		deleteToken    = make([]string, 0)
-		if constant.PlatformIDToClass(platformID) == constant.TerminalPC ||
+		kickToken      = make([]string, 0)
-			constant.PlatformIDToClass(token.PlatformID) == constant.TerminalPC {
+		adminToken     = make([]string, 0)
-			return false
+		unkickTerminal = ""
+	)
+
+	for plfID, tks := range tokens {
+		for k, v := range tks {
+			_, err := tokenverify.GetClaimFromToken(k, authverify.Secret(a.accessSecret))
+			if err != nil || v != constant.NormalToken {
+				deleteToken = append(deleteToken, k)
+			} else {
+				if plfID != constant.AdminPlatformID {
+					loginTokenMap[plfID] = append(loginTokenMap[plfID], k)
+				} else {
+					adminToken = append(adminToken, k)
+				}
+			}
 		}
-		return true
-	case constant.AllLoginButSameTermKick:
-		if platformID == token.PlatformID {
-			return true
-		}
-		return false
-	default:
-		return false
 	}
+
+	switch a.multiLogin.Policy {
+	case constant.DefalutNotKick:
+		for plt, ts := range loginTokenMap {
+			l := len(ts)
+			if platformID == plt {
+				l++
+			}
+			limit := a.multiLogin.MaxNumOneEnd
+			if l > limit {
+				kickToken = append(kickToken, ts[:l-limit]...)
+			}
+		}
+	case constant.AllLoginButSameTermKick:
+		for plt, ts := range loginTokenMap {
+			kickToken = append(kickToken, ts[:len(ts)-1]...)
+			if plt == platformID {
+				kickToken = append(kickToken, ts[len(ts)-1])
+			}
+		}
+	case constant.SingleTerminalLogin:
+		for _, ts := range loginTokenMap {
+			kickToken = append(kickToken, ts...)
+		}
+	case constant.WebAndOther:
+		unkickTerminal = constant.WebPlatformStr
+		fallthrough
+	case constant.PCAndOther:
+		if unkickTerminal == "" {
+			unkickTerminal = constant.TerminalPC
+		}
+		if constant.PlatformIDToClass(platformID) != unkickTerminal {
+			for plt, ts := range loginTokenMap {
+				if constant.PlatformIDToClass(plt) != unkickTerminal {
+					kickToken = append(kickToken, ts...)
+				}
+			}
+		} else {
+			var (
+				preKick   []string
+				isReserve = true
+			)
+			for plt, ts := range loginTokenMap {
+				if constant.PlatformIDToClass(plt) != unkickTerminal {
+					// Keep a token from another end
+					if isReserve {
+						isReserve = false
+						kickToken = append(kickToken, ts[:len(ts)-1]...)
+						preKick = append(preKick, ts[len(ts)-1])
+						continue
+					} else {
+						// Prioritize keeping Android
+						if plt == constant.AndroidPlatformID {
+							kickToken = append(kickToken, preKick...)
+							kickToken = append(kickToken, ts[:len(ts)-1]...)
+						} else {
+							kickToken = append(kickToken, ts...)
+						}
+					}
+				}
+			}
+		}
+	case constant.PcMobileAndWeb:
+		var (
+			reserved = make(map[string]bool)
+		)
+
+		for plt, ts := range loginTokenMap {
+			if constant.PlatformIDToClass(plt) == constant.PlatformIDToClass(platformID) {
+				kickToken = append(kickToken, ts...)
+			} else {
+				if !reserved[constant.PlatformIDToClass(plt)] {
+					reserved[constant.PlatformIDToClass(plt)] = true
+					kickToken = append(kickToken, ts[:len(ts)-1]...)
+					continue
+				} else {
+					kickToken = append(kickToken, ts...)
+				}
+			}
+		}
+
+	case constant.Customize:
+		if a.multiLogin.CustomizeLoginNum[platformID] <= 0 {
+			return nil, nil, errs.New("Do not allow login on this end").Wrap()
+		}
+		for plt, ts := range loginTokenMap {
+			l := len(ts)
+			if platformID == plt {
+				l++
+			}
+			// a.multiLogin.CustomizeLoginNum[platformID] must > 0
+			limit := min(a.multiLogin.CustomizeLoginNum[plt], a.multiLogin.MaxNumOneEnd)
+			if l > limit {
+				kickToken = append(kickToken, ts[:l-limit]...)
+			}
+		}
+	default:
+		return nil, nil, errs.New("unknown multiLogin policy").Wrap()
+	}
+
+	var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
+	if a.multiLogin.Policy == constant.Customize {
+		adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID]
+	}
+	l := len(adminToken)
+	if platformID == constant.AdminPlatformID {
+		l++
+	}
+	if l > adminTokenMaxNum {
+		kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
+	}
+	return deleteToken, kickToken, nil
 }

pkg/common/storage/controller/msg_transfer.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"context"
+	"github.com/openimsdk/protocol/constant"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@@ -83,6 +84,9 @@ func (db *msgTransferDatabase) BatchInsertChat2DB(ctx context.Context, conversat
 				IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
 			}
 		}
+		if msg.Status == constant.MsgStatusSending {
+			msg.Status = constant.MsgStatusSendSuccess
+		}
 		msgs[i] = &model.MsgDataModel{
 			SendID:           msg.SendID,
 			RecvID:           msg.RecvID,
@@ -254,7 +258,7 @@ func (db *msgTransferDatabase) BatchInsertChat2Cache(ctx context.Context, conver
 
 func (db *msgTransferDatabase) setHasReadSeqs(ctx context.Context, conversationID string, userSeqMap map[string]int64) error {
 	for userID, seq := range userSeqMap {
-		if err := db.seqUser.SetUserReadSeq(ctx, conversationID, userID, seq); err != nil {
+		if err := db.seqUser.SetUserReadSeqToDB(ctx, conversationID, userID, seq); err != nil {
 			return err
 		}
 	}

pkg/rpcclient/auth.go

@@ -16,8 +16,8 @@ package rpcclient
 
 import (
 	"context"
+
 	"github.com/openimsdk/protocol/auth"
-	pbAuth "github.com/openimsdk/protocol/auth"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/system/program"
 	"google.golang.org/grpc"
@@ -38,8 +38,8 @@ type Auth struct {
 	discov discovery.SvcDiscoveryRegistry
 }
 
-func (a *Auth) ParseToken(ctx context.Context, token string) (*pbAuth.ParseTokenResp, error) {
+func (a *Auth) ParseToken(ctx context.Context, token string) (*auth.ParseTokenResp, error) {
-	req := pbAuth.ParseTokenReq{
+	req := auth.ParseTokenReq{
 		Token: token,
 	}
 	resp, err := a.Client.ParseToken(ctx, &req)
@@ -49,8 +49,8 @@ func (a *Auth) ParseToken(ctx context.Context, token string) (*pbAuth.ParseToken
 	return resp, err
 }
 
-func (a *Auth) InvalidateToken(ctx context.Context, preservedToken, userID string, platformID int) (*pbAuth.InvalidateTokenResp, error) {
+func (a *Auth) InvalidateToken(ctx context.Context, preservedToken, userID string, platformID int) (*auth.InvalidateTokenResp, error) {
-	req := pbAuth.InvalidateTokenReq{
+	req := auth.InvalidateTokenReq{
 		PreservedToken: preservedToken,
 		UserID:         userID,
 		PlatformID:     int32(platformID),
@@ -61,3 +61,14 @@ func (a *Auth) InvalidateToken(ctx context.Context, preservedToken, userID strin
 	}
 	return resp, err
 }
+
+func (a *Auth) KickTokens(ctx context.Context, tokens []string) (*auth.KickTokensResp, error) {
+	req := auth.KickTokensReq{
+		Tokens: tokens,
+	}
+	resp, err := a.Client.KickTokens(ctx, &req)
+	if err != nil {
+		return nil, err
+	}
+	return resp, err
+}