Merge remote-tracking branch 'origin/del' into del

internal/api/third/fcm_update_token.go

@@ -1,6 +1,7 @@
 package apiThird
 
 import (
+	api "Open_IM/pkg/base_info"
 	"Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/common/token_verify"
@@ -10,18 +11,10 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
-/**
- * FCM第三方上报Token
- */
-type FcmUpdateTokenReq struct {
-	OperationID string `json:"operationID"`
-	Platform    int    `json:"platform" binding:"required,min=1,max=2"` //only for ios + android
-	FcmToken    string `json:"fcmToken"`
-}
-
 func FcmUpdateToken(c *gin.Context) {
 	var (
-		req FcmUpdateTokenReq
+		req  api.FcmUpdateTokenReq
+		resp api.FcmUpdateTokenResp
 	)
 	if err := c.Bind(&req); err != nil {
 		log.NewError("0", utils.GetSelfFuncName(), "BindJSON failed ", err.Error())
@@ -34,7 +27,9 @@ func FcmUpdateToken(c *gin.Context) {
 	if !ok {
 		errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
 		log.NewError(req.OperationID, errMsg)
-		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
+		resp.ErrCode = 500
+		resp.ErrMsg = errMsg
+		c.JSON(http.StatusInternalServerError, resp)
 		return
 	}
 	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), req, UserId)
@@ -43,10 +38,12 @@ func FcmUpdateToken(c *gin.Context) {
 	if err != nil {
 		errMsg := req.OperationID + " " + "SetFcmToken failed " + err.Error() + " token:" + c.Request.Header.Get("token")
 		log.NewError(req.OperationID, errMsg)
-		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
+		resp.ErrCode = 500
+		resp.ErrMsg = errMsg
+		c.JSON(http.StatusInternalServerError, resp)
 		return
 	}
 	//逻辑处理完毕
-	c.JSON(http.StatusOK, gin.H{"errCode": 0, "errMsg": ""})
+	c.JSON(http.StatusOK, resp)
 	return
 }

internal/msg_gateway/gate/ws_server.go

@@ -12,9 +12,10 @@ import (
 	"bytes"
 	"context"
 	"encoding/gob"
+	"strings"
+
 	go_redis "github.com/go-redis/redis/v8"
 	"github.com/pkg/errors"
-	"strings"
 
 	//"gopkg.in/errgo.v2/errors"
 	"net/http"
@@ -133,6 +134,7 @@ func (ws *WServer) MultiTerminalLoginRemoteChecker(userID string, platformID int
 		resp, err := client.MultiTerminalLoginCheck(context.Background(), req)
 		if err != nil {
 			log.Error(operationID, "MultiTerminalLoginCheck failed ", err.Error())
+			continue
 		}
 		if resp.ErrCode != 0 {
 			log.Error(operationID, "MultiTerminalLoginCheck errCode, errMsg: ", resp.ErrCode, resp.ErrMsg)
@@ -237,7 +239,7 @@ func (ws *WServer) MultiTerminalLoginChecker(uid string, platformID int, newConn
 					log.NewError(operationID, utils.GetSelfFuncName(), "callbackUserOffline failed", callbackResp)
 				}
 			} else {
-				log.NewWarn(operationID, "normal uid-conn  ", uid, platformID, oldConnMap[platformID])
+				log.Debug(operationID, "normal uid-conn  ", uid, platformID, oldConnMap[platformID])
 			}
 
 		} else {

internal/msg_transfer/logic/online_history_msg_handler.go

@@ -407,7 +407,7 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerG
 					rwLock.Unlock()
 					split := 1000
 					triggerID = utils.OperationIDGenerator()
-					log.NewWarn(triggerID, "timer trigger msg consumer start", len(ccMsg))
+					log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
 					for i := 0; i < len(ccMsg)/split; i++ {
 						//log.Debug()
 						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
@@ -419,9 +419,8 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerG
 					}
 					//sess.MarkMessage(ccMsg[len(cMsg)-1], "")
 
-					log.NewWarn(triggerID, "timer trigger msg consumer end", len(cMsg))
+					log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
 				}
-
 			}
 		}
 

internal/push/logic/push_handler.go

@@ -13,6 +13,7 @@ import (
 	"Open_IM/pkg/common/log"
 	pbChat "Open_IM/pkg/proto/msg"
 	pbPush "Open_IM/pkg/proto/push"
+	"Open_IM/pkg/utils"
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
 )
@@ -43,6 +44,11 @@ func (ms *PushConsumerHandler) handleMs2PsChat(msg []byte) {
 		MsgData:      msgFromMQ.MsgData,
 		PushToUserID: msgFromMQ.PushToUserID,
 	}
+	sec := msgFromMQ.MsgData.SendTime / 1000
+	nowSec := utils.GetCurrentTimestampBySecond()
+	if nowSec-sec > 10 {
+		return
+	}
 	switch msgFromMQ.MsgData.SessionType {
 	case constant.SuperGroupChatType:
 		MsgToSuperGroupUser(pbData)
@@ -59,6 +65,7 @@ func (ms *PushConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	for msg := range claim.Messages() {
 		log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
 		ms.msgHandle[msg.Topic](msg.Value)
+		sess.MarkMessage(msg, "")
 	}
 	return nil
 }

internal/rpc/group/group.go

@@ -1414,7 +1414,11 @@ func (s *groupServer) GetGroups(_ context.Context, req *pbGroup.GetGroupsReq) (*
 		log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetGroups error", err.Error())
 		return resp, http.WrapError(constant.ErrDB)
 	}
-
+	resp.GroupNum, err = imdb.GetGroupsCountNum(db.Group{})
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetGroupsCountNum error", err.Error())
+		return resp, http.WrapError(constant.ErrDB)
+	}
 	resp.Pagination.PageNumber = req.Pagination.PageNumber
 	resp.Pagination.ShowNumber = req.Pagination.ShowNumber
 	for _, v := range groups {

internal/rpc/organization/organization.go

@@ -305,31 +305,8 @@ func (s *organizationServer) CreateOrganizationUser(ctx context.Context, req *rp
 }
 
 func (s *organizationServer) UpdateOrganizationUser(ctx context.Context, req *rpc.UpdateOrganizationUserReq) (*rpc.UpdateOrganizationUserResp, error) {
-	authReq := &pbAuth.UserRegisterReq{UserInfo: &open_im_sdk.UserInfo{}}
-	utils.CopyStructFields(authReq.UserInfo, req.OrganizationUser)
-	authReq.OperationID = req.OperationID
-	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImAuthName, req.OperationID)
-	if etcdConn == nil {
-		errMsg := req.OperationID + "getcdv3.GetConn == nil"
-		log.NewError(req.OperationID, errMsg)
-		return &rpc.UpdateOrganizationUserResp{ErrCode: constant.ErrInternal.ErrCode, ErrMsg: errMsg}, nil
-	}
-	client := pbAuth.NewAuthClient(etcdConn)
-
-	reply, err := client.UserRegister(context.Background(), authReq)
-	if err != nil {
-		errMsg := "UserRegister failed " + err.Error()
-		log.NewError(req.OperationID, errMsg)
-		return &rpc.UpdateOrganizationUserResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: errMsg}, nil
-	}
-	if reply.CommonResp.ErrCode != 0 {
-		errMsg := "UserRegister failed " + reply.CommonResp.ErrMsg
-		log.NewError(req.OperationID, errMsg)
-		return &rpc.UpdateOrganizationUserResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: errMsg}, nil
-	}
-
 	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), " rpc args ", req.String())
-	if !token_verify.IsManagerUserID(req.OpUserID) {
+	if !token_verify.IsManagerUserID(req.OpUserID) && req.OpUserID != req.OrganizationUser.UserID {
 		errMsg := req.OperationID + " " + req.OpUserID + " is not app manager"
 		log.Error(req.OperationID, errMsg)
 		return &rpc.UpdateOrganizationUserResp{ErrCode: constant.ErrAccess.ErrCode, ErrMsg: errMsg}, nil
@@ -342,7 +319,7 @@ func (s *organizationServer) UpdateOrganizationUser(ctx context.Context, req *rp
 	}
 
 	log.Debug(req.OperationID, "src ", *req.OrganizationUser, "dst ", organizationUser)
-	err = imdb.UpdateOrganizationUser(&organizationUser, nil)
+	err := imdb.UpdateOrganizationUser(&organizationUser, nil)
 	if err != nil {
 		errMsg := req.OperationID + " " + "CreateOrganizationUser failed " + err.Error()
 		log.Error(req.OperationID, errMsg, organizationUser)

internal/rpc/user/user.go

@@ -12,6 +12,7 @@ import (
 	"Open_IM/pkg/common/token_verify"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	pbFriend "Open_IM/pkg/proto/friend"
+	pbOrganization "Open_IM/pkg/proto/organization"
 	sdkws "Open_IM/pkg/proto/sdk_ws"
 	pbUser "Open_IM/pkg/proto/user"
 	"Open_IM/pkg/utils"
@@ -432,21 +433,33 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbUser.UpdateUserI
 	if req.UserInfo.Nickname != "" {
 		go s.SyncJoinedGroupMemberNickname(req.UserInfo.UserID, req.UserInfo.Nickname, oldNickname, req.OperationID, req.OpUserID)
 	}
-	//updateUserInfoToCacheReq := &cache.UpdateUserInfoToCacheReq{
-	//	OperationID:  req.OperationID,
-	//	UserInfoList: []*sdkws.UserInfo{req.UserInfo},
-	//}
-	//cacheEtcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImCacheName)
-	//cacheClient := cache.NewCacheClient(cacheEtcdConn)
-	//resp, err := cacheClient.UpdateUserInfoToCache(context.Background(), updateUserInfoToCacheReq)
-	//if err != nil {
-	//	log.NewError(req.OperationID, utils.GetSelfFuncName(), err.Error(), updateUserInfoToCacheReq.String())
-	//	return &pbUser.UpdateUserInfoResp{CommonResp: &pbUser.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: err.Error()}}, nil
-	//}
-	//if resp.CommonResp.ErrCode != 0 {
-	//	log.NewError(req.OperationID, utils.GetSelfFuncName(), resp.String())
-	//	return &pbUser.UpdateUserInfoResp{CommonResp: &pbUser.CommonResp{ErrCode: constant.ErrServer.ErrCode, ErrMsg: resp.CommonResp.ErrMsg}}, nil
-	//}
+
+	etcdConn = getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOrganizationName, req.OperationID)
+	clientOrg := pbOrganization.NewOrganizationClient(etcdConn)
+	out, err := clientOrg.UpdateOrganizationUser(context.Background(), &pbOrganization.UpdateOrganizationUserReq{
+		OrganizationUser: &sdkws.OrganizationUser{
+			UserID:      req.UserInfo.UserID,
+			Nickname:    req.UserInfo.Nickname,
+			EnglishName: req.UserInfo.Nickname,
+			FaceURL:     req.UserInfo.FaceURL,
+			Gender:      req.UserInfo.Gender,
+			Mobile:      req.UserInfo.PhoneNumber,
+			Telephone:   req.UserInfo.PhoneNumber,
+			Birth:       req.UserInfo.Birth,
+			Email:       req.UserInfo.Email,
+			Ex:          req.UserInfo.Ex,
+		},
+		OperationID: req.OperationID,
+		OpUserID:    req.OpUserID,
+	})
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "UpdateOrganizationUser failed", err.Error())
+	} else {
+		if out.ErrCode != 0 {
+			log.NewError(req.OperationID, utils.GetSelfFuncName(), "grpc resp: ", out)
+		}
+	}
+
 	return &pbUser.UpdateUserInfoResp{CommonResp: &pbUser.CommonResp{}}, nil
 }
 func (s *userServer) SetGlobalRecvMessageOpt(ctx context.Context, req *pbUser.SetGlobalRecvMessageOptReq) (*pbUser.SetGlobalRecvMessageOptResp, error) {