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

cmd/open_im_api/main.go

@@ -92,7 +92,8 @@ func main() {
 	}
 	superGroupRouterGroup := r.Group("/super_group")
 	{
-		superGroupRouterGroup.POST("/get_joined_super_group_list", group.GetJoinedSuperGroupList)
+		superGroupRouterGroup.POST("/get_joined_group_list", group.GetJoinedSuperGroupList)
+		superGroupRouterGroup.POST("/get_groups_info", group.GetSuperGroupsInfo)
 	}
 	//certificate
 	authRouterGroup := r.Group("/auth")

internal/api/group/super_group.go

@@ -5,6 +5,7 @@ import (
 	api "Open_IM/pkg/base_info"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/token_verify"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	rpc "Open_IM/pkg/proto/group"
 	"Open_IM/pkg/utils"
@@ -15,14 +16,21 @@ import (
 )
 
 func GetJoinedSuperGroupList(c *gin.Context) {
-	req := api.GetJoinedSuperGroupReq{}
+	req := api.GetJoinedSuperGroupListReq{}
 	if err := c.BindJSON(&req); err != nil {
 		log.NewError("0", "BindJSON failed ", err.Error())
 		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
 		return
 	}
 	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req)
-	reqPb := rpc.GetJoinedSuperGroupListReq{OperationID: req.OperationID}
+	ok, opUserID, errInfo := token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
+	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})
+		return
+	}
+	reqPb := rpc.GetJoinedSuperGroupListReq{OperationID: req.OperationID, OpUserID: opUserID, UserID: req.FromUserID}
 	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
 	client := rpc.NewGroupClient(etcdConn)
 	rpcResp, err := client.GetJoinedSuperGroupList(context.Background(), &reqPb)
@@ -31,8 +39,38 @@ func GetJoinedSuperGroupList(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
 		return
 	}
-	GroupListResp := api.GetJoinedGroupListResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupList}
+	GroupListResp := api.GetJoinedSuperGroupListResp{GetJoinedGroupListResp: api.GetJoinedGroupListResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupList}}
 	GroupListResp.Data = jsonData.JsonDataList(GroupListResp.GroupInfoList)
 	log.NewInfo(req.OperationID, "GetJoinedGroupList api return ", GroupListResp)
 	c.JSON(http.StatusOK, GroupListResp)
 }
+
+func GetSuperGroupsInfo(c *gin.Context) {
+	req := api.GetSuperGroupsInfoReq{}
+	if err := c.BindJSON(&req); err != nil {
+		log.NewError("0", "BindJSON failed ", err.Error())
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req)
+	ok, opUserID, errInfo := token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
+	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})
+		return
+	}
+	reqPb := rpc.GetSuperGroupsInfoReq{OperationID: req.OperationID, OpUserID: opUserID, GroupIDList: req.GroupIDList}
+	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
+	client := rpc.NewGroupClient(etcdConn)
+	rpcResp, err := client.GetSuperGroupsInfo(context.Background(), &reqPb)
+	if err != nil {
+		log.NewError(req.OperationID, "InviteUserToGroup failed ", err.Error(), reqPb.String())
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
+		return
+	}
+	resp := api.GetSuperGroupsInfoResp{GetGroupInfoResp: api.GetGroupInfoResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupInfoList}}
+	resp.Data = jsonData.JsonDataList(resp.GroupInfoList)
+	log.NewInfo(req.OperationID, "GetGroupsInfo api return ", resp)
+	c.JSON(http.StatusOK, resp)
+}

internal/push/logic/push_rpc_server.go

@@ -2,6 +2,7 @@ package logic
 
 import (
 	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	"Open_IM/pkg/proto/push"
@@ -63,7 +64,12 @@ func (r *RPCServer) run() {
 }
 func (r *RPCServer) PushMsg(_ context.Context, pbData *pbPush.PushMsgReq) (*pbPush.PushMsgResp, error) {
 	//Call push module to send message to the user
+	switch pbData.MsgData.SessionType {
+	case constant.SuperGroupChatType:
+		MsgToSuperGroupUser(pbData)
+	default:
 		MsgToUser(pbData)
+	}
 	return &pbPush.PushMsgResp{
 		ResultCode: 0,
 	}, nil

internal/push/logic/push_to_client.go

@@ -132,6 +132,104 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
 
 	}
 }
+func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
+	return
+	var wsResult []*pbRelay.SingleMsgToUser
+	isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush)
+	log.Debug(pushMsg.OperationID, "Get msg from msg_transfer And push msg", pushMsg.String())
+	if len(grpcCons) == 0 {
+		log.NewWarn(pushMsg.OperationID, "first GetConn4Unique ")
+		grpcCons = getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
+	}
+	//Online push message
+	log.Debug("test", pushMsg.OperationID, "len  grpc", len(grpcCons), "data", pushMsg.String())
+	for _, v := range grpcCons {
+		msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v)
+		reply, err := msgClient.OnlinePushMsg(context.Background(), &pbRelay.OnlinePushMsgReq{OperationID: pushMsg.OperationID, MsgData: pushMsg.MsgData, PushToUserID: pushMsg.PushToUserID})
+		if err != nil {
+			log.NewError("push data to client rpc err", pushMsg.OperationID, "err", err)
+			continue
+		}
+		if reply != nil && reply.Resp != nil {
+			wsResult = append(wsResult, reply.Resp...)
+		}
+	}
+	log.NewInfo(pushMsg.OperationID, "push_result", wsResult, "sendData", pushMsg.MsgData)
+	successCount++
+	if isOfflinePush && pushMsg.PushToUserID != pushMsg.MsgData.SendID {
+		for _, v := range wsResult {
+			if v.ResultCode == 0 {
+				if utils.IsContainInt32(v.RecvPlatFormID, pushTerminal) {
+					break
+				}
+				continue
+			}
+			if utils.IsContainInt32(v.RecvPlatFormID, pushTerminal) {
+				//Use offline push messaging
+				var UIDList []string
+				UIDList = append(UIDList, v.RecvID)
+				customContent := OpenIMContent{
+					SessionType: int(pushMsg.MsgData.SessionType),
+					From:        pushMsg.MsgData.SendID,
+					To:          pushMsg.MsgData.RecvID,
+					Seq:         pushMsg.MsgData.Seq,
+				}
+				bCustomContent, _ := json.Marshal(customContent)
+				jsonCustomContent := string(bCustomContent)
+				var content string
+				if pushMsg.MsgData.OfflinePushInfo != nil {
+					content = pushMsg.MsgData.OfflinePushInfo.Title
+
+				} else {
+					switch pushMsg.MsgData.ContentType {
+					case constant.Text:
+						content = constant.ContentType2PushContent[constant.Text]
+					case constant.Picture:
+						content = constant.ContentType2PushContent[constant.Picture]
+					case constant.Voice:
+						content = constant.ContentType2PushContent[constant.Voice]
+					case constant.Video:
+						content = constant.ContentType2PushContent[constant.Video]
+					case constant.File:
+						content = constant.ContentType2PushContent[constant.File]
+					case constant.AtText:
+						a := AtContent{}
+						_ = utils.JsonStringToStruct(string(pushMsg.MsgData.Content), &a)
+						if utils.IsContain(v.RecvID, a.AtUserList) {
+							content = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
+						} else {
+							content = constant.ContentType2PushContent[constant.GroupMsg]
+						}
+					default:
+						content = constant.ContentType2PushContent[constant.Common]
+					}
+				}
+				callbackResp := callbackOfflinePush(pushMsg.OperationID, UIDList[0], pushMsg.MsgData.OfflinePushInfo, v.RecvPlatFormID)
+				log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), "offline callback Resp")
+				if callbackResp.ErrCode != 0 {
+					log.NewError(pushMsg.OperationID, utils.GetSelfFuncName(), "callbackOfflinePush result: ", callbackResp)
+				}
+				if callbackResp.ActionCode != constant.ActionAllow {
+					log.NewDebug(pushMsg.OperationID, utils.GetSelfFuncName(), "offlinePush stop")
+					break
+				}
+
+				if offlinePusher == nil {
+					offlinePusher = jpush.JPushClient
+				}
+				pushResult, err := offlinePusher.Push(UIDList, content, jsonCustomContent, pushMsg.OperationID)
+				if err != nil {
+					log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error())
+				} else {
+					log.NewDebug(pushMsg.OperationID, "offline push return result is ", pushResult, pushMsg.MsgData)
+				}
+				break
+			}
+
+		}
+
+	}
+}
 
 //func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) {
 //	m.MsgID = rpcChat.GetMsgID(m.SendID)

internal/rpc/group/super_group.go

@@ -5,6 +5,7 @@ import (
 	"Open_IM/pkg/common/db"
 	imdb "Open_IM/pkg/common/db/mysql_model/im_mysql_model"
 	"Open_IM/pkg/common/log"
+	cp "Open_IM/pkg/common/utils"
 	pbGroup "Open_IM/pkg/proto/group"
 	commonPb "Open_IM/pkg/proto/sdk_ws"
 	"Open_IM/pkg/utils"
@@ -42,3 +43,22 @@ func (s *groupServer) GetJoinedSuperGroupList(ctx context.Context, req *pbGroup.
 	log.NewError(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
 	return resp, nil
 }
+
+func (s *groupServer) GetSuperGroupsInfo(_ context.Context, req *pbGroup.GetSuperGroupsInfoReq) (resp *pbGroup.GetSuperGroupsInfoResp, err error) {
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
+	resp = &pbGroup.GetSuperGroupsInfoResp{CommonResp: &pbGroup.CommonResp{}}
+	groupsInfoList := make([]*commonPb.GroupInfo, 0)
+	for _, groupID := range req.GroupIDList {
+		groupInfoFromMysql, err := imdb.GetGroupInfoByGroupID(groupID)
+		if err != nil {
+			log.NewError(req.OperationID, "GetGroupInfoByGroupID failed ", err.Error(), groupID)
+			continue
+		}
+		var groupInfo commonPb.GroupInfo
+		cp.GroupDBCopyOpenIM(&groupInfo, groupInfoFromMysql)
+		groupsInfoList = append(groupsInfoList, &groupInfo)
+	}
+	resp.GroupInfoList = groupsInfoList
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "resp: ", resp.String())
+	return resp, nil
+}

pkg/base_info/super_group.go

@@ -1,10 +1,10 @@
 package base_info
 
-type GetJoinedSuperGroupReq struct {
+type GetJoinedSuperGroupListReq struct {
 	GetJoinedGroupListReq
 }
 
-type GetJoinedSuperGroupResp struct {
+type GetJoinedSuperGroupListResp struct {
 	GetJoinedGroupListResp
 }
 

pkg/common/db/mongoModel.go

@@ -985,11 +985,14 @@ func (d *DataBases) CreateSuperGroup(groupID string, initMemberIDList []string,
 	opts := &options.UpdateOptions{
 		Upsert: &upsert,
 	}
-	_, err = c.UpdateMany(sCtx, bson.M{"user_id": bson.M{"$in": initMemberIDList}}, bson.M{"$addToSet": bson.M{"group_id_list": groupID}}, opts)
+	c = d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cUserToSuperGroup)
+	for _, userID := range initMemberIDList {
+		_, err = c.UpdateOne(sCtx, bson.M{"user_id": userID}, bson.M{"$addToSet": bson.M{"group_id_list": groupID}}, opts)
 		if err != nil {
 			session.AbortTransaction(ctx)
 			return utils.Wrap(err, "transaction failed")
 		}
+	}
 	session.CommitTransaction(ctx)
 	return err
 }
@@ -1026,11 +1029,17 @@ func (d *DataBases) AddUserToSuperGroup(groupID string, userIDList []string) err
 			UserID: v,
 		})
 	}
-	_, err = c.UpdateMany(sCtx, users, bson.M{"$addToSet": bson.M{"group_id_list": groupID}})
+	upsert := true
+	opts := &options.UpdateOptions{
+		Upsert: &upsert,
+	}
+	for _, userID := range userIDList {
+		_, err = c.UpdateOne(sCtx, bson.M{"user_id": userID}, bson.M{"$addToSet": bson.M{"group_id_list": groupID}}, opts)
 		if err != nil {
 			session.AbortTransaction(ctx)
 			return utils.Wrap(err, "transaction failed")
 		}
+	}
 	session.CommitTransaction(ctx)
 	return err
 }
@@ -1097,7 +1106,7 @@ func (d *DataBases) RemoveGroupFromUser(ctx, sCtx context.Context, groupID strin
 		})
 	}
 	c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cUserToSuperGroup)
-	_, err := c.UpdateOne(sCtx, bson.M{"user_id": groupID}, bson.M{"$pull": bson.M{"group_id_list": groupID}})
+	_, err := c.UpdateOne(sCtx, bson.M{"user_id": bson.M{"$in": userID}}, bson.M{"$pull": bson.M{"group_id_list": groupID}})
 	if err != nil {
 		return utils.Wrap(err, "UpdateOne transaction failed")
 	}

pkg/proto/group/group.proto

@@ -401,6 +401,16 @@ message GetJoinedSuperGroupListResp {
   repeated server_api_params.GroupInfo GroupList = 3;
 }
 
+message GetSuperGroupsInfoReq {
+  repeated string GroupIDList = 1;
+  string OperationID = 2;
+  string OpUserID = 3; //No verification permission
+}
+
+message GetSuperGroupsInfoResp {
+  CommonResp commonResp = 1;
+  repeated server_api_params.GroupInfo GroupInfoList = 3;
+}
 
 service group{
   rpc createGroup(CreateGroupReq) returns(CreateGroupResp);
@@ -438,6 +448,7 @@ service group{
   rpc SetGroupMemberNickname(SetGroupMemberNicknameReq) returns (SetGroupMemberNicknameResp);
 
   rpc GetJoinedSuperGroupList(GetJoinedSuperGroupListReq) returns (GetJoinedSuperGroupListResp);
+  rpc GetSuperGroupsInfo(GetSuperGroupsInfoReq) returns (GetSuperGroupsInfoResp);
 }