Merge branch 'superGroup' of github.com:OpenIMSDK/Open-IM-Server into superGroup

config/config.yaml

@@ -261,6 +261,7 @@ messageverify:
 iospush:
   pushSound: "xxx"
   badgeCount: true
+  production: false
 
 callback:
   # callback url 需要自行更换callback url

go.mod

@@ -45,6 +45,7 @@ require (
 	google.golang.org/grpc v1.45.0
 	google.golang.org/protobuf v1.27.1
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
+	gopkg.in/errgo.v2 v2.1.0
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
 	gopkg.in/ini.v1 v1.66.2 // indirect
 	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22

go.sum

@@ -562,6 +562,8 @@ gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8=
+gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df h1:n7WqCuqOuCbNr617RXOY0AWRXxgwEyPp2z+p0+hgMuE=
 gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df/go.mod h1:LRQQ+SO6ZHR7tOkpBDuZnXENFzX8qRjMDMyPD6BRkCw=

internal/api/auth/auth.go

@@ -100,7 +100,7 @@ func UserToken(c *gin.Context) {
 	client := rpc.NewAuthClient(etcdConn)
 	reply, err := client.UserToken(context.Background(), req)
 	if err != nil {
-		errMsg := req.OperationID + " UserToken failed " + err.Error() + req.String()
+		errMsg := req.OperationID + " UserToken failed " + err.Error() + " req: " + req.String()
 		log.NewError(req.OperationID, errMsg)
 		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
 		return

internal/msg_gateway/gate/logic.go

@@ -3,6 +3,7 @@ package gate
 import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	pbChat "Open_IM/pkg/proto/chat"
@@ -170,7 +171,7 @@ func (ws *WServer) sendMsgReq(conn *UserConn, m *Req) {
 	sendMsgAllCountLock.Lock()
 	sendMsgAllCount++
 	sendMsgAllCountLock.Unlock()
-	log.NewInfo(m.OperationID, "Ws call success to sendMsgReq start", m.MsgIncr, m.ReqIdentifier, m.SendID, m.Data)
+	log.NewInfo(m.OperationID, "Ws call success to sendMsgReq start", m.MsgIncr, m.ReqIdentifier, m.SendID)
 
 	nReply := new(pbChat.SendMsgResp)
 	isPass, errCode, errMsg, pData := ws.argsValidate(m, constant.WSSendMsg, m.OperationID)
@@ -181,7 +182,7 @@ func (ws *WServer) sendMsgReq(conn *UserConn, m *Req) {
 			OperationID: m.OperationID,
 			MsgData:     &data,
 		}
-		log.NewInfo(m.OperationID, "Ws call success to sendMsgReq middle", m.ReqIdentifier, m.SendID, m.MsgIncr, data)
+		log.NewInfo(m.OperationID, "Ws call success to sendMsgReq middle", m.ReqIdentifier, m.SendID, m.MsgIncr, data.String())
 		etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName, m.OperationID)
 		if etcdConn == nil {
 			errMsg := m.OperationID + "getcdv3.GetConn == nil"
@@ -331,3 +332,19 @@ func (ws *WServer) sendErrMsg(conn *UserConn, errCode int32, errMsg string, reqI
 	}
 	ws.sendMsg(conn, mReply)
 }
+
+func SetTokenKicked(userID string, platformID int, operationID string) {
+	m, err := db.DB.GetTokenMapByUidPid(userID, constant.PlatformIDToName(platformID))
+	if err != nil {
+		log.Error(operationID, "GetTokenMapByUidPid failed ", err.Error(), userID, constant.PlatformIDToName(platformID))
+		return
+	}
+	for k, _ := range m {
+		m[k] = constant.KickedToken
+	}
+	err = db.DB.SetTokenMapByUidPid(userID, platformID, m)
+	if err != nil {
+		log.Error(operationID, "SetTokenMapByUidPid failed ", err.Error(), userID, constant.PlatformIDToName(platformID))
+		return
+	}
+}

internal/msg_gateway/gate/rpc_server.go

@@ -292,11 +292,13 @@ func (r *RPCServer) KickUserOffline(_ context.Context, req *pbRelay.KickUserOffl
 		oldConnMap := ws.getUserAllCons(v)
 		if conn, ok := oldConnMap[int(req.PlatformID)]; ok { // user->map[platform->conn]
 			ws.sendKickMsg(conn, &UserConn{})
+			conn.Close()
 		}
+		SetTokenKicked(v, int(req.PlatformID), req.OperationID)
 	}
 	return &pbRelay.KickUserOfflineResp{}, nil
-
 }
+
 func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.OnlinePushMsgReq, RecvPlatForm int, RecvID string) (ResultCode int64) {
 	err := ws.writeMsg(conn, websocket.BinaryMessage, bMsg)
 	if err != nil {

internal/msg_gateway/gate/ws_server.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	"encoding/gob"
 	go_redis "github.com/go-redis/redis/v8"
+	"gopkg.in/errgo.v2/errors"
 	"net/http"
 	"sync"
 	"time"
@@ -289,6 +290,24 @@ func (ws *WServer) headerCheck(w http.ResponseWriter, r *http.Request, operation
 	query := r.URL.Query()
 	if len(query["token"]) != 0 && len(query["sendID"]) != 0 && len(query["platformID"]) != 0 {
 		if ok, err, msg := token_verify.WsVerifyToken(query["token"][0], query["sendID"][0], query["platformID"][0], operationID); !ok {
+			switch errors.Cause(err) {
+			case constant.ErrTokenExpired:
+				status = int(constant.ErrTokenExpired.ErrCode)
+			case constant.ErrTokenInvalid:
+				status = int(constant.ErrTokenInvalid.ErrCode)
+			case constant.ErrTokenMalformed:
+				status = int(constant.ErrTokenMalformed.ErrCode)
+			case constant.ErrTokenNotValidYet:
+				status = int(constant.ErrTokenNotValidYet.ErrCode)
+			case constant.ErrTokenUnknown:
+				status = int(constant.ErrTokenUnknown.ErrCode)
+			case constant.ErrTokenKicked:
+				status = int(constant.ErrTokenKicked.ErrCode)
+			case constant.ErrTokenDifferentPlatformID:
+				status = int(constant.ErrTokenDifferentPlatformID.ErrCode)
+			case constant.ErrTokenDifferentUserID:
+				status = int(constant.ErrTokenDifferentUserID.ErrCode)
+			}
 			log.Error(operationID, "Token verify failed ", "query ", query, msg, err.Error())
 			w.Header().Set("Sec-Websocket-Version", "13")
 			w.Header().Set("ws_err_msg", err.Error())
@@ -299,6 +318,7 @@ func (ws *WServer) headerCheck(w http.ResponseWriter, r *http.Request, operation
 			return true
 		}
 	} else {
+		status = int(constant.ErrArgs.ErrCode)
 		log.Error(operationID, "Args err ", "query ", query)
 		w.Header().Set("Sec-Websocket-Version", "13")
 		errMsg := "args err, need token, sendID, platformID"

internal/push/jpush/push.go

@@ -51,7 +51,7 @@ func (j *JPush) Push(accounts []string, alert, detailContent, operationID string
 	var me requestBody.Message
 	me.SetMsgContent(detailContent)
 	var o requestBody.Options
-	o.SetApnsProduction(false)
+	o.SetApnsProduction(config.Config.IOSPush.Production)
 	var po requestBody.PushObj
 	po.SetPlatform(&pf)
 	po.SetAudience(&au)

internal/rpc/group/group.go

@@ -208,8 +208,11 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
 		if req.GroupInfo.GroupType != constant.SuperGroup {
 			chat.GroupCreatedNotification(req.OperationID, req.OpUserID, groupId, okUserIDList)
 		} else {
-			chat.SuperGroupNotification(req.OperationID, req.OpUserID, groupId)
+			go func() {
-
+				for _, v := range okUserIDList {
+					chat.SuperGroupNotification(req.OperationID, v, v)
+				}
+			}()
 		}
 		return resp, nil
 	} else {
@@ -406,7 +409,11 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
 	if groupInfo.GroupType != constant.SuperGroup {
 		chat.MemberInvitedNotification(req.OperationID, req.GroupID, req.OpUserID, req.Reason, okUserIDList)
 	} else {
-		chat.SuperGroupNotification(req.OperationID, req.OpUserID, req.GroupID)
+		go func() {
+			for _, v := range req.InvitedUserIDList {
+				chat.SuperGroupNotification(req.OperationID, v, v)
+			}
+		}()
 	}
 
 	log.NewInfo(req.OperationID, "InviteUserToGroup rpc return ")
@@ -620,7 +627,12 @@ func (s *groupServer) KickGroupMember(ctx context.Context, req *pbGroup.KickGrou
 	if groupInfo.GroupType != constant.SuperGroup {
 		chat.MemberKickedNotification(req, okUserIDList)
 	} else {
-		chat.SuperGroupNotification(req.OperationID, req.OpUserID, req.GroupID)
+		go func() {
+			for _, v := range req.KickedUserIDList {
+				chat.SuperGroupNotification(req.OperationID, v, v)
+			}
+		}()
+
 	}
 	log.NewInfo(req.OperationID, "GetGroupMemberList rpc return ", resp.String())
 	return &resp, nil
@@ -928,7 +940,7 @@ func (s *groupServer) QuitGroup(ctx context.Context, req *pbGroup.QuitGroupReq)
 	if groupInfo.GroupType != constant.SuperGroup {
 		chat.MemberQuitNotification(req)
 	} else {
-		chat.SuperGroupNotification(req.OperationID, req.OpUserID, req.GroupID)
+		chat.SuperGroupNotification(req.OperationID, req.OpUserID, req.OpUserID)
 	}
 
 	log.NewInfo(req.OperationID, "rpc QuitGroup return ", pbGroup.QuitGroupResp{CommonResp: &pbGroup.CommonResp{ErrCode: 0, ErrMsg: ""}})

internal/rpc/msg/super_group_notification.go

@@ -9,7 +9,7 @@ import (
 	//"github.com/golang/protobuf/proto"
 )
 
-func SuperGroupNotification(operationID, sendID, groupID string) {
+func SuperGroupNotification(operationID, sendID, recvID string) {
 
 	//var tips sdk.TipsComm
 	//var err error
@@ -21,10 +21,10 @@ func SuperGroupNotification(operationID, sendID, groupID string) {
 	//tips.JsonDetail, _ = marshaler.MarshalToString(m)
 	n := &NotificationMsg{
 		SendID:      sendID,
-		RecvID:      groupID,
+		RecvID:      recvID,
 		MsgFrom:     constant.SysMsgType,
 		ContentType: constant.SuperGroupUpdateNotification,
-		SessionType: constant.SuperGroupChatType,
+		SessionType: constant.SingleChatType,
 		OperationID: operationID,
 	}
 	//n.Content, err = proto.Marshal(&tips)

internal/utils/jwt_token_test.go

@@ -38,7 +38,7 @@ func Test_CreateToken(t *testing.T) {
 	platform := int32(1)
 	now := time.Now().Unix()
 
-	tokenString, expiresAt, err := token_verify.CreateToken(uid, platform)
+	tokenString, expiresAt, err := token_verify.CreateToken(uid, int(platform))
 
 	assert.NotEmpty(t, tokenString)
 	assert.Equal(t, expiresAt, 604800+now)
@@ -48,7 +48,7 @@ func Test_CreateToken(t *testing.T) {
 func Test_VerifyToken(t *testing.T) {
 	uid := "1"
 	platform := int32(1)
-	tokenString, _, _ := token_verify.CreateToken(uid, platform)
+	tokenString, _, _ := token_verify.CreateToken(uid, int(platform))
 	result, _ := token_verify.VerifyToken(tokenString, uid)
 	assert.True(t, result)
 	result, _ = token_verify.VerifyToken(tokenString, "2")
@@ -58,7 +58,7 @@ func Test_VerifyToken(t *testing.T) {
 func Test_ParseRedisInterfaceToken(t *testing.T) {
 	uid := "1"
 	platform := int32(1)
-	tokenString, _, _ := token_verify.CreateToken(uid, platform)
+	tokenString, _, _ := token_verify.CreateToken(uid, int(platform))
 
 	claims, err := token_verify.ParseRedisInterfaceToken([]uint8(tokenString))
 	assert.Nil(t, err)
@@ -66,7 +66,7 @@ func Test_ParseRedisInterfaceToken(t *testing.T) {
 
 	// timeout
 	config.Config.TokenPolicy.AccessExpire = -80
-	tokenString, _, _ = token_verify.CreateToken(uid, platform)
+	tokenString, _, _ = token_verify.CreateToken(uid, int(platform))
 	claims, err = token_verify.ParseRedisInterfaceToken([]uint8(tokenString))
 	assert.Equal(t, err, constant.ExpiredToken)
 	assert.Nil(t, claims)
@@ -75,7 +75,7 @@ func Test_ParseRedisInterfaceToken(t *testing.T) {
 func Test_ParseToken(t *testing.T) {
 	uid := "1"
 	platform := int32(1)
-	tokenString, _, _ := token_verify.CreateToken(uid, platform)
+	tokenString, _, _ := token_verify.CreateToken(uid, int(platform))
 	claims, err := token_verify.ParseToken(tokenString, "")
 	if err == nil {
 		assert.Equal(t, claims.UID, uid)

pkg/common/config/config.go

@@ -238,6 +238,7 @@ type config struct {
 	IOSPush struct {
 		PushSound  string `yaml:"pushSound"`
 		BadgeCount bool   `yaml:"badgeCount"`
+		Production bool   `yaml:"production"`
 	}
 
 	Callback struct {

pkg/common/constant/constant.go

@@ -283,6 +283,10 @@ func GroupIsBanPrivateChat(status int32) bool {
 	return true
 }
 
+const (
+	TokenKicked = 1001
+)
+
 const BigVersion = "v2"
 
 const LogFileName = "OpenIM.log"

pkg/common/constant/error.go

@@ -48,6 +48,8 @@ var (
 	ErrTokenNotValidYet         = ErrInfo{704, TokenNotValidYetMsg.Error()}
 	ErrTokenUnknown             = ErrInfo{705, TokenUnknownMsg.Error()}
 	ErrTokenKicked              = ErrInfo{706, TokenUserKickedMsg.Error()}
+	ErrTokenDifferentPlatformID = ErrInfo{707, TokenDifferentPlatformIDMsg.Error()}
+	ErrTokenDifferentUserID     = ErrInfo{708, TokenDifferentUserIDMsg.Error()}
 
 	ErrAccess                = ErrInfo{ErrCode: 801, ErrMsg: AccessMsg.Error()}
 	ErrDB                    = ErrInfo{ErrCode: 802, ErrMsg: DBMsg.Error()}
@@ -67,6 +69,8 @@ var (
 	TokenMalformedMsg           = errors.New("that's not even a token")
 	TokenUnknownMsg             = errors.New("couldn't handle this token")
 	TokenUserKickedMsg          = errors.New("user has been kicked")
+	TokenDifferentPlatformIDMsg = errors.New("different platformID")
+	TokenDifferentUserIDMsg     = errors.New("different userID")
 	AccessMsg                   = errors.New("no permission")
 	StatusMsg                   = errors.New("status is abnormal")
 	DBMsg                       = errors.New("db failed")

pkg/common/token_verify/jwt_token.go

@@ -188,7 +188,7 @@ func ParseToken(tokensString, operationID string) (claims *Claims, err error) {
 		return nil, utils.Wrap(&constant.ErrTokenInvalid, "get token from redis err")
 	}
 	if m == nil {
-		log.NewError(operationID, "get token from redis err", "m is nil", tokensString)
+		log.NewError(operationID, "get token from redis err, not in redis ", "m is nil", tokensString)
 		return nil, utils.Wrap(&constant.ErrTokenInvalid, "get token from redis err")
 	}
 	if v, ok := m[tokensString]; ok {
@@ -196,13 +196,9 @@ func ParseToken(tokensString, operationID string) (claims *Claims, err error) {
 		case constant.NormalToken:
 			log.NewDebug(operationID, "this is normal return", claims)
 			return claims, nil
-		case constant.InValidToken:
-			return nil, utils.Wrap(&constant.ErrTokenInvalid, "")
 		case constant.KickedToken:
 			log.Error(operationID, "this token has been kicked by other same terminal ", constant.ErrTokenKicked)
 			return nil, utils.Wrap(&constant.ErrTokenKicked, "this token has been kicked by other same terminal ")
-		case constant.ExpiredToken:
-			return nil, utils.Wrap(&constant.ErrTokenExpired, "")
 		default:
 			return nil, utils.Wrap(&constant.ErrTokenUnknown, "")
 		}
@@ -244,20 +240,21 @@ func VerifyToken(token, uid string) (bool, error) {
 	log.NewDebug("", claims.UID, claims.Platform)
 	return true, nil
 }
+
 func WsVerifyToken(token, uid string, platformID string, operationID string) (bool, error, string) {
 	argMsg := "token: " + token + " operationID: " + operationID + " userID: " + uid + " platformID: " + platformID
 	claims, err := ParseToken(token, operationID)
 	if err != nil {
-		errMsg := "parse token err " + argMsg
+		errMsg := "parse token err " + err.Error() + argMsg
 		return false, utils.Wrap(err, errMsg), errMsg
 	}
 	if claims.UID != uid {
 		errMsg := " uid is not same to token uid " + " claims.UID " + claims.UID + argMsg
-		return false, utils.Wrap(&constant.ErrTokenUnknown, errMsg), errMsg
+		return false, utils.Wrap(&constant.ErrTokenDifferentUserID, errMsg), errMsg
 	}
 	if claims.Platform != constant.PlatformIDToName(utils.StringToInt(platformID)) {
 		errMsg := " platform is not same to token platform " + argMsg + "claims platformID " + claims.Platform
-		return false, utils.Wrap(&constant.ErrTokenUnknown, errMsg), errMsg
+		return false, utils.Wrap(&constant.ErrTokenDifferentPlatformID, errMsg), errMsg
 	}
 	log.NewDebug(operationID, utils.GetSelfFuncName(), " check ok ", claims.UID, uid, claims.Platform)
 	return true, nil, ""