blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-24 18:36:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

group

Browse Source
This commit is contained in:
withchao 2023-02-10 19:17:33 +08:00
parent 4b3d965783
commit 09c6fbdfe4
10 changed files with 415 additions and 410 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
internal/common/check/user.go
View File

@ -6,6 +6,32 @@ import (
"errors"
)
func GetUsersInfo(ctx context.Context, args ...interface{}) ([]*sdkws.UserInfo, error) {
return nil, errors.New("TODO:GetUserInfo")
//func GetUsersInfo(ctx context.Context, args ...interface{}) ([]*sdkws.UserInfo, error) {
// return nil, errors.New("TODO:GetUserInfo")
//}
func NewUserCheck() *UserCheck {
return &UserCheck{}
}
type UserCheck struct{}
func (u *UserCheck) GetUsersInfos(ctx context.Context, userIDs []string, complete bool) ([]*sdkws.UserInfo, error) {
return nil, errors.New("todo")
}
func (u *UserCheck) GetUsersInfoMap(ctx context.Context, userIDs []string, complete bool) (map[string]*sdkws.UserInfo, error) {
return nil, errors.New("todo")
}
func (u *UserCheck) GetPublicUserInfo(ctx context.Context, userID string) (*sdkws.PublicUserInfo, error) {
return nil, errors.New("todo")
}
func (u *UserCheck) GetPublicUserInfos(ctx context.Context, userIDs []string, complete bool) ([]*sdkws.PublicUserInfo, error) {
return nil, errors.New("todo")
}
func (u *UserCheck) GetPublicUserInfoMap(ctx context.Context, userIDs []string, complete bool) (map[string]*sdkws.PublicUserInfo, error) {
return nil, errors.New("todo")
}

103
internal/rpc/group/callback.go
View File

@ -1,42 +1,56 @@
package group
import (
cbApi "Open_IM/pkg/callback_struct"
"Open_IM/pkg/apistruct"
"Open_IM/pkg/callbackstruct"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db/table/relation"
"Open_IM/pkg/common/http"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/tracelog"
pbGroup "Open_IM/pkg/proto/group"
"Open_IM/pkg/proto/group"
"Open_IM/pkg/utils"
"context"
"google.golang.org/protobuf/types/known/wrapperspb"
)
func callbackBeforeCreateGroup(ctx context.Context, req *pbGroup.CreateGroupReq) (err error) {
defer func() {
tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "req", req)
}()
func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (err error) {
if !config.Config.Callback.CallbackBeforeCreateGroup.Enable {
return nil
}
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), req.String())
commonCallbackReq := &cbApi.CallbackBeforeCreateGroupReq{
defer func() {
tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "req", req)
}()
operationID := tracelog.GetOperationID(ctx)
commonCallbackReq := &callbackstruct.CallbackBeforeCreateGroupReq{
CallbackCommand: constant.CallbackBeforeCreateGroupCommand,
OperationID: req.OperationID,
OperationID: operationID,
GroupInfo: *req.GroupInfo,
InitMemberList: req.InitMemberList,
}
callbackResp := cbApi.CommonCallbackResp{OperationID: req.OperationID}
resp := &cbApi.CallbackBeforeCreateGroupResp{
CommonCallbackResp: &callbackResp,
commonCallbackReq.InitMemberList = append(commonCallbackReq.InitMemberList, &apistruct.GroupAddMemberInfo{
UserID: req.OwnerUserID,
RoleLevel: constant.GroupOwner,
})
for _, userID := range req.AdminUserIDs {
commonCallbackReq.InitMemberList = append(commonCallbackReq.InitMemberList, &apistruct.GroupAddMemberInfo{
UserID: userID,
RoleLevel: constant.GroupAdmin,
})
}
//utils.CopyStructFields(req, msg.MsgData)
defer log.NewDebug(req.OperationID, utils.GetSelfFuncName(), commonCallbackReq, *resp)
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeCreateGroupCommand, commonCallbackReq,
resp, config.Config.Callback.CallbackBeforeCreateGroup)
if err == nil {
for _, userID := range req.AdminUserIDs {
commonCallbackReq.InitMemberList = append(commonCallbackReq.InitMemberList, &apistruct.GroupAddMemberInfo{
UserID: userID,
RoleLevel: constant.GroupOrdinaryUsers,
})
}
resp := &callbackstruct.CallbackBeforeCreateGroupResp{
CommonCallbackResp: &callbackstruct.CommonCallbackResp{OperationID: operationID},
}
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeCreateGroupCommand, commonCallbackReq, resp, config.Config.Callback.CallbackBeforeCreateGroup)
if err != nil {
return err
}
if resp.GroupID != nil {
req.GroupInfo.GroupID = *resp.GroupID
}
@ -73,20 +87,19 @@ func callbackBeforeCreateGroup(ctx context.Context, req *pbGroup.CreateGroupReq)
if resp.LookMemberInfo != nil {
req.GroupInfo.LookMemberInfo = *resp.LookMemberInfo
}
}
return err
return nil
}
func CallbackBeforeMemberJoinGroup(ctx context.Context, operationID string, groupMember *relation.GroupMemberModel, groupEx string) (err error) {
defer func() {
tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "groupMember", *groupMember, "groupEx", groupEx)
}()
callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
func CallbackBeforeMemberJoinGroup(ctx context.Context, groupMember *relation.GroupMemberModel, groupEx string) (err error) {
if !config.Config.Callback.CallbackBeforeMemberJoinGroup.Enable {
return nil
}
log.NewDebug(operationID, "args: ", *groupMember)
callbackReq := cbApi.CallbackBeforeMemberJoinGroupReq{
defer func() {
tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "groupMember", *groupMember, "groupEx", groupEx)
}()
operationID := tracelog.GetOperationID(ctx)
callbackResp := callbackstruct.CommonCallbackResp{OperationID: operationID}
callbackReq := callbackstruct.CallbackBeforeMemberJoinGroupReq{
CallbackCommand: constant.CallbackBeforeMemberJoinGroupCommand,
OperationID: operationID,
GroupID: groupMember.GroupID,
@ -94,12 +107,14 @@ func CallbackBeforeMemberJoinGroup(ctx context.Context, operationID string, grou
Ex: groupMember.Ex,
GroupEx: groupEx,
}
resp := &cbApi.CallbackBeforeMemberJoinGroupResp{
resp := &callbackstruct.CallbackBeforeMemberJoinGroupResp{
CommonCallbackResp: &callbackResp,
}
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeMemberJoinGroupCommand, callbackReq,
resp, config.Config.Callback.CallbackBeforeMemberJoinGroup)
if err == nil {
if err != nil {
return err
}
if resp.MuteEndTime != nil {
groupMember.MuteEndTime = utils.UnixSecondToTime(*resp.MuteEndTime)
}
@ -115,21 +130,21 @@ func CallbackBeforeMemberJoinGroup(ctx context.Context, operationID string, grou
if resp.RoleLevel != nil {
groupMember.RoleLevel = *resp.RoleLevel
}
}
return err
return nil
}
func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *pbGroup.SetGroupMemberInfo) (err error) {
defer func() {
tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "req", *req)
}()
callbackResp := cbApi.CommonCallbackResp{OperationID: req.OperationID}
func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMemberInfo) (err error) {
if !config.Config.Callback.CallbackBeforeSetGroupMemberInfo.Enable {
return nil
}
callbackReq := cbApi.CallbackBeforeSetGroupMemberInfoReq{
defer func() {
tracelog.SetCtxInfo(ctx, utils.GetFuncName(1), err, "req", *req)
}()
operationID := tracelog.GetOperationID(ctx)
callbackResp := callbackstruct.CommonCallbackResp{OperationID: operationID}
callbackReq := callbackstruct.CallbackBeforeSetGroupMemberInfoReq{
CallbackCommand: constant.CallbackBeforeSetGroupMemberInfoCommand,
OperationID: req.OperationID,
OperationID: operationID,
GroupID: req.GroupID,
UserID: req.UserID,
}
@ -145,12 +160,13 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *pbGroup.SetGroup
if req.Ex != nil {
callbackReq.Ex = req.Ex.Value
}
resp := &cbApi.CallbackBeforeSetGroupMemberInfoResp{
resp := &callbackstruct.CallbackBeforeSetGroupMemberInfoResp{
CommonCallbackResp: &callbackResp,
}
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetGroupMemberInfoCommand, callbackReq,
resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo.CallbackTimeOut, &config.Config.Callback.CallbackBeforeSetGroupMemberInfo.CallbackFailedContinue)
if err == nil {
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetGroupMemberInfoCommand, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo)
if err != nil {
return err
}
if resp.FaceURL != nil {
req.FaceURL = &wrapperspb.StringValue{Value: *resp.FaceURL}
}
@ -163,6 +179,5 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *pbGroup.SetGroup
if resp.Ex != nil {
req.Ex = &wrapperspb.StringValue{Value: *resp.Ex}
}
}
return err
}

94
internal/rpc/group/g.go
View File

@ -1,94 +0,0 @@
package group
import (
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/tracelog"
pbConversation "Open_IM/pkg/proto/conversation"
sdkws "Open_IM/pkg/proto/sdkws"
"Open_IM/pkg/utils"
"context"
"errors"
"math/big"
"strconv"
"strings"
"time"
)
func GetPublicUserInfoOne(ctx context.Context, userID string) (*sdkws.PublicUserInfo, error) {
return nil, errors.New("todo")
}
func GetUsersInfo(ctx context.Context, userIDs []string) ([]*sdkws.UserInfo, error) {
return nil, errors.New("todo")
}
func GetUserInfoMap(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error) {
users, err := GetUsersInfo(ctx, userIDs)
if err != nil {
return nil, err
}
return utils.SliceToMap(users, func(e *sdkws.UserInfo) string {
return e.UserID
}), nil
}
func GetPublicUserInfo(ctx context.Context, userIDs []string) ([]*sdkws.PublicUserInfo, error) {
return nil, errors.New("todo")
}
func GetPublicUserInfoMap(ctx context.Context, userIDs []string) (map[string]*sdkws.PublicUserInfo, error) {
users, err := GetPublicUserInfo(ctx, userIDs)
if err != nil {
return nil, err
}
return utils.SliceToMap(users, func(e *sdkws.PublicUserInfo) string {
return e.UserID
}), nil
}
func GetUsername(ctx context.Context, userIDs []string) (map[string]string, error) {
if len(userIDs) == 0 {
return map[string]string{}, nil
}
users, err := GetPublicUserInfo(ctx, userIDs)
if err != nil {
return nil, err
}
if ids := utils.Single(userIDs, utils.Slice(users, func(e *sdkws.PublicUserInfo) string {
return e.UserID
})); len(ids) > 0 {
return nil, constant.ErrUserIDNotFound.Wrap(strings.Join(ids, ","))
}
return utils.SliceToMapAny(users, func(e *sdkws.PublicUserInfo) (string, string) {
return e.UserID, e.Nickname
}), nil
}
func GroupNotification(ctx context.Context, groupID string) {
var conversationReq pbConversation.ModifyConversationFieldReq
conversation := pbConversation.Conversation{
OwnerUserID: tracelog.GetOpUserID(ctx),
ConversationID: utils.GetConversationIDBySessionType(groupID, constant.GroupChatType),
ConversationType: constant.GroupChatType,
GroupID: groupID,
}
conversationReq.Conversation = &conversation
conversationReq.OperationID = tracelog.GetOperationID(ctx)
conversationReq.FieldType = constant.FieldGroupAtType
conversation.GroupAtType = constant.GroupNotification
conversationReq.UserIDList = cacheResp.UserIDList
_, err = pbConversation.NewConversationClient(s.etcdConn.GetConn("", config.Config.RpcRegisterName.OpenImConversationName)).ModifyConversationField(ctx, &conversationReq)
tracelog.SetCtxInfo(ctx, "ModifyConversationField", err, "req", &conversationReq, "resp", conversationReply)
}
func genGroupID(ctx context.Context, groupID string) string {
if groupID != "" {
return groupID
}
groupID = utils.Md5(tracelog.GetOperationID(ctx) + strconv.FormatInt(time.Now().UnixNano(), 10))
bi := big.NewInt(0)
bi.SetString(groupID[0:8], 16)
groupID = bi.String()
return groupID
}

142
internal/rpc/group/group.go
View File

@ -1,6 +1,7 @@
package group
import (
"Open_IM/internal/common/check"
"Open_IM/internal/common/network"
chat "Open_IM/internal/rpc/msg"
"Open_IM/pkg/common/config"
@ -22,6 +23,9 @@ import (
"context"
"fmt"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"gorm.io/gorm"
"math/big"
"math/rand"
"net"
"strconv"
"strings"
@ -38,6 +42,7 @@ type groupServer struct {
zkAddr []string
GroupInterface controller.GroupInterface
registerCenter discoveryRegistry.SvcDiscoveryRegistry
user *check.UserCheck
}
func NewGroupServer(port int) *groupServer {
@ -72,8 +77,8 @@ func NewGroupServer(port int) *groupServer {
}
//conns, err := g.registerCenter.GetConns(config.Config.RpcRegisterName.OpenImConversationName)
g.GroupInterface = controller.NewGroupInterface(mysql.GormConn(), redis.GetClient(), mongo.GetClient())
g.user = check.NewUserCheck()
return &g
}
@ -132,6 +137,69 @@ func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error
return nil
}
func (s *groupServer) GetUsernameMap(ctx context.Context, userIDs []string, complete bool) (map[string]string, error) {
if len(userIDs) == 0 {
return map[string]string{}, nil
}
users, err := s.user.GetPublicUserInfos(ctx, userIDs, complete)
if err != nil {
return nil, err
}
return utils.SliceToMapAny(users, func(e *open_im_sdk.PublicUserInfo) (string, string) {
return e.UserID, e.Nickname
}), nil
}
func (s *groupServer) GroupNotification(ctx context.Context, groupID string) {
// todo 群公告修改通知
//var conversationReq pbConversation.ModifyConversationFieldReq
//conversation := pbConversation.Conversation{
// OwnerUserID: tracelog.GetOpUserID(ctx),
// ConversationID: utils.GetConversationIDBySessionType(groupID, constant.GroupChatType),
// ConversationType: constant.GroupChatType,
// GroupID: groupID,
//}
//conversationReq.Conversation = &conversation
//conversationReq.FieldType = constant.FieldGroupAtType
//conversation.GroupAtType = constant.GroupNotification
//conversationReq.UserIDList = userIDs
//_, err := pbConversation.NewConversationClient(s.etcdConn.GetConn("", config.Config.RpcRegisterName.OpenImConversationName)).ModifyConversationField(ctx, &conversationReq)
//tracelog.SetCtxInfo(ctx, "ModifyConversationField", err, "req", &conversationReq, "resp", conversationReply)
}
func (s *groupServer) IsNotFound(err error) bool {
return utils.Unwrap(err) == gorm.ErrRecordNotFound
}
func (s *groupServer) GenGroupID(ctx context.Context, groupID *string) error {
if *groupID != "" {
_, err := s.GroupInterface.TakeGroup(ctx, *groupID)
if err == nil {
return constant.ErrGroupIDExisted.Wrap("group id existed " + *groupID)
} else if s.IsNotFound(err) {
return nil
} else {
return err
}
}
for i := 0; i < 10; i++ {
id := utils.Md5(strings.Join([]string{tracelog.GetOperationID(ctx), strconv.FormatInt(time.Now().UnixNano(), 10), strconv.Itoa(rand.Int())}, ",;,"))
bi := big.NewInt(0)
bi.SetString(id[0:8], 16)
id = bi.String()
_, err := s.GroupInterface.TakeGroup(ctx, id)
if err == nil {
continue
} else if s.IsNotFound(err) {
*groupID = id
return nil
} else {
return err
}
}
return constant.ErrData.Wrap("group id gen error")
}
func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupReq) (*pbGroup.CreateGroupResp, error) {
resp := &pbGroup.CreateGroupResp{GroupInfo: &open_im_sdk.GroupInfo{}}
if err := tokenverify.CheckAccessV3(ctx, req.OwnerUserID); err != nil {
@ -144,19 +212,18 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
if utils.Duplicate(userIDs) {
return nil, constant.ErrArgs.Wrap("group member repeated")
}
userMap, err := GetUserInfoMap(ctx, userIDs)
userMap, err := s.user.GetUsersInfoMap(ctx, userIDs, true)
if err != nil {
return nil, err
}
if ids := utils.Single(userIDs, utils.Keys(userMap)); len(ids) > 0 {
return nil, constant.ErrUserIDNotFound.Wrap(strings.Join(ids, ","))
}
if err := callbackBeforeCreateGroup(ctx, req); err != nil {
if err := CallbackBeforeCreateGroup(ctx, req); err != nil {
return nil, err
}
var groupMembers []*relationTb.GroupMemberModel
group := PbToDBGroupInfo(req.GroupInfo)
group.GroupID = genGroupID(ctx, req.GroupInfo.GroupID)
if err := s.GenGroupID(ctx, &group.GroupID); err != nil {
return nil, err
}
joinGroup := func(userID string, roleLevel int32) error {
groupMember := PbToDbGroupMember(userMap[userID])
groupMember.Nickname = ""
@ -165,7 +232,7 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
groupMember.OperatorUserID = tracelog.GetOpUserID(ctx)
groupMember.JoinSource = constant.JoinByInvitation
groupMember.InviterUserID = tracelog.GetOpUserID(ctx)
if err := CallbackBeforeMemberJoinGroup(ctx, tracelog.GetOperationID(ctx), groupMember, group.Ex); err != nil {
if err := CallbackBeforeMemberJoinGroup(ctx, groupMember, group.Ex); err != nil {
return err
}
groupMembers = append(groupMembers, groupMember)
@ -271,13 +338,10 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
if ids := utils.Single(req.InvitedUserIDs, utils.Keys(memberMap)); len(ids) > 0 {
return nil, constant.ErrArgs.Wrap("user in group " + strings.Join(ids, ","))
}
userMap, err := GetUserInfoMap(ctx, req.InvitedUserIDs)
userMap, err := s.user.GetUsersInfoMap(ctx, req.InvitedUserIDs, true)
if err != nil {
return nil, err
}
if ids := utils.Single(req.InvitedUserIDs, utils.Keys(userMap)); len(ids) > 0 {
return nil, constant.ErrArgs.Wrap("user not found " + strings.Join(ids, ","))
}
if group.NeedVerification == constant.AllNeedVerification {
if !tokenverify.IsAppManagerUid(ctx) {
opUserID := tracelog.GetOpUserID(ctx)
@ -328,7 +392,7 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
member.OperatorUserID = opUserID
member.InviterUserID = opUserID
member.JoinSource = constant.JoinByInvitation
if err := CallbackBeforeMemberJoinGroup(ctx, tracelog.GetOperationID(ctx), member, group.Ex); err != nil {
if err := CallbackBeforeMemberJoinGroup(ctx, member, group.Ex); err != nil {
return nil, err
}
groupMembers = append(groupMembers, member)
@ -354,9 +418,9 @@ func (s *groupServer) GetGroupAllMember(ctx context.Context, req *pbGroup.GetGro
if err != nil {
return nil, err
}
nameMap, err := GetUsername(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
return e.UserID, e.Nickname == ""
}))
}), true)
if err != nil {
return nil, err
}
@ -376,9 +440,9 @@ func (s *groupServer) GetGroupMemberList(ctx context.Context, req *pbGroup.GetGr
return nil, err
}
resp.Total = total
nameMap, err := GetUsername(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
return e.UserID, e.Nickname == ""
}))
}), true)
if err != nil {
return nil, err
}
@ -470,9 +534,9 @@ func (s *groupServer) GetGroupMembersInfo(ctx context.Context, req *pbGroup.GetG
if err != nil {
return nil, err
}
nameMap, err := GetUsername(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
return e.UserID, e.Nickname == ""
}))
}), true)
if err != nil {
return nil, err
}
@ -505,13 +569,10 @@ func (s *groupServer) GetGroupApplicationList(ctx context.Context, req *pbGroup.
}
userIDs = utils.Distinct(userIDs)
groupIDs = utils.Distinct(groupIDs)
userMap, err := GetPublicUserInfoMap(ctx, userIDs)
userMap, err := s.user.GetPublicUserInfoMap(ctx, userIDs, true)
if err != nil {
return nil, err
}
if ids := utils.Single(utils.Keys(userMap), userIDs); len(ids) > 0 {
return nil, constant.ErrUserIDNotFound.Wrap(strings.Join(ids, ","))
}
groups, err := s.GroupInterface.FindGroup(ctx, utils.Distinct(groupIDs))
if err != nil {
return nil, err
@ -590,19 +651,18 @@ func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbGroup
if groupRequest.HandleResult != 0 {
return nil, constant.ErrArgs.Wrap("group request already processed")
}
if _, err := s.GroupInterface.TakeGroupMember(ctx, req.GroupID, req.FromUserID); err != nil {
if !IsNotFound(err) {
var join bool
if _, err = s.GroupInterface.TakeGroupMember(ctx, req.GroupID, req.FromUserID); err == nil {
join = true // 已经在群里了
} else if !s.IsNotFound(err) {
return nil, err
}
} else {
return nil, constant.ErrArgs.Wrap("already in group")
}
user, err := GetPublicUserInfoOne(ctx, req.FromUserID)
user, err := s.user.GetPublicUserInfo(ctx, req.FromUserID)
if err != nil {
return nil, err
}
var member *relationTb.GroupMemberModel
if req.HandleResult == constant.GroupResponseAgree {
if (!join) && req.HandleResult == constant.GroupResponseAgree {
member = &relationTb.GroupMemberModel{
GroupID: req.GroupID,
UserID: user.UserID,
@ -615,25 +675,27 @@ func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbGroup
OperatorUserID: tracelog.GetOpUserID(ctx),
Ex: groupRequest.Ex,
}
if err = CallbackBeforeMemberJoinGroup(ctx, tracelog.GetOperationID(ctx), member, group.Ex); err != nil {
if err = CallbackBeforeMemberJoinGroup(ctx, member, group.Ex); err != nil {
return nil, err
}
}
if err := s.GroupInterface.HandlerGroupRequest(ctx, req.GroupID, req.FromUserID, req.HandledMsg, req.HandleResult, member); err != nil {
return nil, err
}
if !join {
if req.HandleResult == constant.GroupResponseAgree {
chat.GroupApplicationAcceptedNotification(req)
chat.MemberEnterNotification(req)
chat.MemberEnterNotification(ctx, req)
} else if req.HandleResult == constant.GroupResponseRefuse {
chat.GroupApplicationRejectedNotification(req)
}
}
return resp, nil
}
func (s *groupServer) JoinGroup(ctx context.Context, req *pbGroup.JoinGroupReq) (*pbGroup.JoinGroupResp, error) {
resp := &pbGroup.JoinGroupResp{}
if _, err := GetPublicUserInfoOne(ctx, tracelog.GetOpUserID(ctx)); err != nil {
if _, err := s.user.GetPublicUserInfo(ctx, tracelog.GetOpUserID(ctx)); err != nil {
return nil, err
}
group, err := s.GroupInterface.TakeGroup(ctx, req.GroupID)
@ -657,7 +719,7 @@ func (s *groupServer) JoinGroup(ctx context.Context, req *pbGroup.JoinGroupReq)
groupMember.OperatorUserID = tracelog.GetOpUserID(ctx)
groupMember.JoinSource = constant.JoinByInvitation
groupMember.InviterUserID = tracelog.GetOpUserID(ctx)
if err := CallbackBeforeMemberJoinGroup(ctx, tracelog.GetOperationID(ctx), groupMember, group.Ex); err != nil {
if err := CallbackBeforeMemberJoinGroup(ctx, groupMember, group.Ex); err != nil {
return nil, err
}
if err := s.GroupInterface.CreateGroup(ctx, nil, []*relationTb.GroupMemberModel{groupMember}); err != nil {
@ -732,7 +794,7 @@ func (s *groupServer) SetGroupInfo(ctx context.Context, req *pbGroup.SetGroupInf
}
chat.GroupInfoSetNotification(tracelog.GetOperationID(ctx), tracelog.GetOpUserID(ctx), req.GroupInfoForSet.GroupID, group.GroupName, group.Notification, group.Introduction, group.FaceURL, req.GroupInfoForSet.NeedVerification)
if req.GroupInfoForSet.Notification != "" {
GroupNotification(ctx, group.GroupID)
s.GroupNotification(ctx, group.GroupID)
}
return resp, nil
}
@ -830,9 +892,9 @@ func (s *groupServer) GetGroupMembersCMS(ctx context.Context, req *pbGroup.GetGr
return nil, err
}
resp.Total = total
nameMap, err := GetUsername(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
return e.UserID, e.Nickname == ""
}))
}), true)
if err != nil {
return nil, err
}
@ -847,7 +909,7 @@ func (s *groupServer) GetGroupMembersCMS(ctx context.Context, req *pbGroup.GetGr
func (s *groupServer) GetUserReqApplicationList(ctx context.Context, req *pbGroup.GetUserReqApplicationListReq) (*pbGroup.GetUserReqApplicationListResp, error) {
resp := &pbGroup.GetUserReqApplicationListResp{}
user, err := GetPublicUserInfoOne(ctx, req.UserID)
user, err := s.user.GetPublicUserInfo(ctx, req.UserID)
if err != nil {
return nil, err
}
@ -1105,9 +1167,9 @@ func (s *groupServer) GetUserInGroupMembers(ctx context.Context, req *pbGroup.Ge
if err != nil {
return nil, err
}
nameMap, err := GetUsername(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationTb.GroupMemberModel) (string, bool) {
return e.UserID, e.Nickname == ""
}))
}), true)
if err != nil {
return nil, err
}

13
internal/rpc/group/utils.go
View File

@ -1,13 +0,0 @@
package group
import (
"Open_IM/pkg/common/tracelog"
"gorm.io/gorm"
)
func IsNotFound(err error) bool {
if err == nil {
return false
}
return tracelog.Unwrap(err) == gorm.ErrRecordNotFound
}

4
pkg/callbackstruct/group.go
View File

@ -1,7 +1,7 @@
package callbackstruct
import (
"Open_IM/pkg/proto/group"
"Open_IM/pkg/apistruct"
common "Open_IM/pkg/proto/sdkws"
)
@ -9,7 +9,7 @@ type CallbackBeforeCreateGroupReq struct {
CallbackCommand string `json:"callbackCommand"`
OperationID string `json:"operationID"`
common.GroupInfo
InitMemberList []*group.GroupAddMemberInfo `json:"initMemberList"`
InitMemberList []*apistruct.GroupAddMemberInfo `json:"initMemberList"`
}
type CallbackBeforeCreateGroupResp struct {

1
pkg/common/constant/errors.go
View File

@ -13,6 +13,7 @@ var (
ErrUserIDNotFound = &ErrInfo{UserIDNotFoundError, "UserIDNotFoundError", ""}
ErrGroupIDNotFound = &ErrInfo{GroupIDNotFoundError, "GroupIDNotFoundError", ""}
ErrGroupIDExisted = &ErrInfo{GroupIDNotFoundError, "GroupIDExisted", ""} // todo group id 已存在
ErrRecordNotFound = &ErrInfo{RecordNotFoundError, "RecordNotFoundError", ""}

241
pkg/common/db/cache/group.go vendored
View File

@ -1,17 +1,14 @@
package cache
import (
"Open_IM/pkg/common/db/relation"
relationTb "Open_IM/pkg/common/db/table/relation"
"Open_IM/pkg/common/db/unrelation"
"Open_IM/pkg/common/tracelog"
"Open_IM/pkg/utils"
"context"
"encoding/json"
"github.com/dtm-labs/rockscache"
"github.com/go-redis/redis/v8"
"math/big"
"strconv"
"strings"
"time"
)
@ -157,6 +154,27 @@ func (g *GroupCacheRedis) GetGroupMembersHash(ctx context.Context, groupID strin
})
}
func (g *GroupCacheRedis) GetGroupMemberHash1(ctx context.Context, groupIDs []string) (map[string]*relationTb.GroupSimpleUserID, error) {
// todo
mapGroupUserIDs, err := g.groupMember.FindJoinUserID(ctx, groupIDs)
if err != nil {
return nil, err
}
res := make(map[string]*relationTb.GroupSimpleUserID)
for _, groupID := range groupIDs {
userIDs := mapGroupUserIDs[groupID]
users := &relationTb.GroupSimpleUserID{}
if len(userIDs) > 0 {
utils.Sort(userIDs, true)
bi := big.NewInt(0)
bi.SetString(utils.Md5(strings.Join(userIDs, ";"))[0:8], 16)
users.Hash = bi.Uint64()
}
res[groupID] = users
}
return res, nil
}
func (g *GroupCacheRedis) DelGroupMembersHash(ctx context.Context, groupID string) (err error) {
defer func() {
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupID", groupID)
@ -178,111 +196,104 @@ func (g *GroupCacheRedis) DelGroupMemberIDs(ctx context.Context, groupID string)
return g.rcClient.TagAsDeleted(g.getGroupMemberIDsKey(groupID))
}
// JoinedGroups
func (g *GroupCacheRedis) GetJoinedGroupIDs(ctx context.Context, userID string) (joinedGroupIDs []string, err error) {
getJoinedGroupIDList := func() (string, error) {
joinedGroupList, err := relation.GetJoinedGroupIDListByUserID(userID)
if err != nil {
return "", err
}
bytes, err := json.Marshal(joinedGroupList)
if err != nil {
return "", utils.Wrap(err, "")
}
return string(bytes), nil
}
defer func() {
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "userID", userID, "joinedGroupIDs", joinedGroupIDs)
}()
joinedGroupIDListStr, err := g.rcClient.Fetch(g.getJoinedGroupsKey(userID), time.Second*30*60, getJoinedGroupIDList)
if err != nil {
return nil, err
}
err = json.Unmarshal([]byte(joinedGroupIDListStr), &joinedGroupIDs)
return joinedGroupIDs, utils.Wrap(err, "")
}
//// JoinedGroups
//func (g *GroupCacheRedis) GetJoinedGroupIDs(ctx context.Context, userID string) (joinedGroupIDs []string, err error) {
// getJoinedGroupIDList := func() (string, error) {
// joinedGroupList, err := relation.GetJoinedGroupIDListByUserID(userID)
// if err != nil {
// return "", err
// }
// bytes, err := json.Marshal(joinedGroupList)
// if err != nil {
// return "", utils.Wrap(err, "")
// }
// return string(bytes), nil
// }
// defer func() {
// tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "userID", userID, "joinedGroupIDs", joinedGroupIDs)
// }()
// joinedGroupIDListStr, err := g.rcClient.Fetch(g.getJoinedGroupsKey(userID), time.Second*30*60, getJoinedGroupIDList)
// if err != nil {
// return nil, err
// }
// err = json.Unmarshal([]byte(joinedGroupIDListStr), &joinedGroupIDs)
// return joinedGroupIDs, utils.Wrap(err, "")
//}
func (g *GroupCacheRedis) DelJoinedGroupIDs(ctx context.Context, userID string) (err error) {
func (g *GroupCacheRedis) DelJoinedGroupID(ctx context.Context, userID string) (err error) {
defer func() {
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "userID", userID)
}()
return g.rcClient.TagAsDeleted(g.getJoinedGroupsKey(userID))
}
// GetGroupMemberInfo
func (g *GroupCacheRedis) GetGroupMemberInfo(ctx context.Context, groupID, userID string) (groupMember *relation.GroupMember, err error) {
getGroupMemberInfo := func() (string, error) {
groupMemberInfo, err := relation.GetGroupMemberInfoByGroupIDAndUserID(groupID, userID)
if err != nil {
return "", err
}
bytes, err := json.Marshal(groupMemberInfo)
if err != nil {
return "", utils.Wrap(err, "")
}
return string(bytes), nil
}
defer func() {
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupID", groupID, "userID", userID, "groupMember", *groupMember)
}()
groupMemberInfoStr, err := g.rcClient.Fetch(g.getGroupMemberInfoKey(groupID, userID), time.Second*30*60, getGroupMemberInfo)
if err != nil {
return nil, err
}
groupMember = &relation.GroupMember{}
err = json.Unmarshal([]byte(groupMemberInfoStr), groupMember)
return groupMember, utils.Wrap(err, "")
//func (g *GroupCacheRedis) DelJoinedGroupIDs(ctx context.Context, userIDs []string) (err error) {
// defer func() {
// tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "userID", userID)
// }()
// for _, userID := range userIDs {
// if err := g.DelJoinedGroupID(ctx, userID); err != nil {
// return err
// }
// }
// return nil
//}
func (g *GroupCacheRedis) GetGroupMemberInfo(ctx context.Context, groupID, userID string) (groupMember *relationTb.GroupMemberModel, err error) {
return GetCache(ctx, g.rcClient, g.getGroupMemberInfoKey(groupID, userID), g.expireTime, func(ctx context.Context) (*relationTb.GroupMemberModel, error) {
return g.groupMember.Take(ctx, groupID, userID)
})
}
func (g *GroupCacheRedis) GetGroupMembersInfo(ctx context.Context, groupID, userIDs []string) (groupMember *relationTb.GroupMemberModel, err error) {
//func (g *GroupCacheRedis) GetGroupMembersInfo(ctx context.Context, groupID, userIDs []string) (groupMember *relationTb.GroupMemberModel, err error) {
//
// return nil, err
//}
return nil, err
}
func (g *GroupCacheRedis) GetGroupMembersInfo(ctx context.Context, count, offset int32, groupID string) (groupMembers []*relation.GroupMember, err error) {
defer func() {
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "count", count, "offset", offset, "groupID", groupID, "groupMember", groupMembers)
}()
groupMemberIDList, err := g.GetGroupMemberIDs(ctx, groupID)
if err != nil {
return nil, err
}
if count < 0 || offset < 0 {
return nil, nil
}
var groupMemberList []*relation.GroupMember
var start, stop int32
start = offset
stop = offset + count
l := int32(len(groupMemberIDList))
if start > stop {
return nil, nil
}
if start >= l {
return nil, nil
}
if count != 0 {
if stop >= l {
stop = l
}
groupMemberIDList = groupMemberIDList[start:stop]
} else {
if l < 1000 {
stop = l
} else {
stop = 1000
}
groupMemberIDList = groupMemberIDList[start:stop]
}
for _, userID := range groupMemberIDList {
groupMember, err := g.GetGroupMemberInfo(ctx, groupID, userID)
if err != nil {
return
}
groupMembers = append(groupMembers, groupMember)
}
return groupMemberList, nil
}
//func (g *GroupCacheRedis) GetGroupMembersInfo(ctx context.Context, count, offset int32, groupID string) (groupMembers []*relation.GroupMember, err error) {
// defer func() {
// tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "count", count, "offset", offset, "groupID", groupID, "groupMember", groupMembers)
// }()
// groupMemberIDList, err := g.GetGroupMemberIDs(ctx, groupID)
// if err != nil {
// return nil, err
// }
// if count < 0 || offset < 0 {
// return nil, nil
// }
// var groupMemberList []*relation.GroupMember
// var start, stop int32
// start = offset
// stop = offset + count
// l := int32(len(groupMemberIDList))
// if start > stop {
// return nil, nil
// }
// if start >= l {
// return nil, nil
// }
// if count != 0 {
// if stop >= l {
// stop = l
// }
// groupMemberIDList = groupMemberIDList[start:stop]
// } else {
// if l < 1000 {
// stop = l
// } else {
// stop = 1000
// }
// groupMemberIDList = groupMemberIDList[start:stop]
// }
// for _, userID := range groupMemberIDList {
// groupMember, err := g.GetGroupMemberInfo(ctx, groupID, userID)
// if err != nil {
// return
// }
// groupMembers = append(groupMembers, groupMember)
// }
// return groupMemberList, nil
//}
func (g *GroupCacheRedis) DelGroupMemberInfo(ctx context.Context, groupID, userID string) (err error) {
defer func() {
@ -292,23 +303,23 @@ func (g *GroupCacheRedis) DelGroupMemberInfo(ctx context.Context, groupID, userI
}
// groupMemberNum
func (g *GroupCacheRedis) GetGroupMemberNum(ctx context.Context, groupID string) (num int, err error) {
getGroupMemberNum := func() (string, error) {
num, err := relation.GetGroupMemberNumByGroupID(groupID)
if err != nil {
return "", err
}
return strconv.Itoa(int(num)), nil
}
defer func() {
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupID", groupID, "num", num)
}()
groupMember, err := g.rcClient.Fetch(g.getGroupMemberNumKey(groupID), time.Second*30*60, getGroupMemberNum)
if err != nil {
return 0, err
}
return strconv.Atoi(groupMember)
}
//func (g *GroupCacheRedis) GetGroupMemberNum(ctx context.Context, groupID string) (num int, err error) {
// getGroupMemberNum := func() (string, error) {
// num, err := relation.GetGroupMemberNumByGroupID(groupID)
// if err != nil {
// return "", err
// }
// return strconv.Itoa(int(num)), nil
// }
// defer func() {
// tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupID", groupID, "num", num)
// }()
// groupMember, err := g.rcClient.Fetch(g.getGroupMemberNumKey(groupID), time.Second*30*60, getGroupMemberNum)
// if err != nil {
// return 0, err
// }
// return strconv.Atoi(groupMember)
//}
func (g *GroupCacheRedis) DelGroupMemberNum(ctx context.Context, groupID string) (err error) {
defer func() {

44
pkg/common/db/controller/group.go
View File

@ -15,8 +15,6 @@ import (
"github.com/go-redis/redis/v8"
"go.mongodb.org/mongo-driver/mongo"
"gorm.io/gorm"
"math/big"
"strings"
)
//type GroupInterface GroupDataBaseInterface
@ -248,7 +246,7 @@ type GroupDataBase struct {
func (g *GroupDataBase) delGroupMemberCache(ctx context.Context, groupID string, userIDs []string) error {
for _, userID := range userIDs {
if err := g.cache.DelJoinedGroupIDs(ctx, userID); err != nil {
if err := g.cache.DelJoinedGroupID(ctx, userID); err != nil {
return err
}
if err := g.cache.DelJoinedSuperGroupIDs(ctx, userID); err != nil {
@ -272,21 +270,24 @@ func (g *GroupDataBase) FindGroupMemberUserID(ctx context.Context, groupID strin
}
func (g *GroupDataBase) CreateGroup(ctx context.Context, groups []*relationTb.GroupModel, groupMembers []*relationTb.GroupMemberModel) error {
if len(groups) > 0 && len(groupMembers) > 0 {
return g.db.Transaction(func(tx *gorm.DB) error {
if len(groups) > 0 {
if err := g.groupDB.Create(ctx, groups, tx); err != nil {
return err
}
return g.groupMemberDB.Create(ctx, groupMembers, tx)
})
}
if len(groups) > 0 {
return g.groupDB.Create(ctx, groups)
}
if len(groupMembers) > 0 {
return g.groupMemberDB.Create(ctx, groupMembers)
if err := g.groupMemberDB.Create(ctx, groupMembers, tx); err != nil {
return err
}
//if err := g.cache.DelJoinedGroupIDs(ctx, utils.Slice(groupMembers, func(e *relationTb.GroupMemberModel) string {
// return e.UserID
//})); err != nil {
// return err
//}
}
return nil
})
}
func (g *GroupDataBase) TakeGroup(ctx context.Context, groupID string) (group *relationTb.GroupModel, err error) {
@ -337,12 +338,11 @@ func (g *GroupDataBase) TakeGroupMember(ctx context.Context, groupID string, use
}
func (g *GroupDataBase) TakeGroupOwner(ctx context.Context, groupID string) (*relationTb.GroupMemberModel, error) {
return g.groupMemberDB.TakeOwner(ctx, groupID)
return g.groupMemberDB.TakeOwner(ctx, groupID) // todo cache group owner
}
func (g *GroupDataBase) FindGroupMember(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32) ([]*relationTb.GroupMemberModel, error) {
//g.cache.GetGroupMembersInfo()
return g.groupMemberDB.Find(ctx, groupIDs, userIDs, roleLevels)
return g.groupMemberDB.Find(ctx, groupIDs, userIDs, roleLevels) // todo cache group find
}
func (g *GroupDataBase) PageGroupMember(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32, pageNumber, showNumber int32) (uint32, []*relationTb.GroupMemberModel, error) {
@ -383,23 +383,7 @@ func (g *GroupDataBase) DeleteGroupMember(ctx context.Context, groupID string, u
}
func (g *GroupDataBase) MapGroupMemberUserID(ctx context.Context, groupIDs []string) (map[string]*relationTb.GroupSimpleUserID, error) {
mapGroupUserIDs, err := g.groupMemberDB.FindJoinUserID(ctx, groupIDs)
if err != nil {
return nil, err
}
res := make(map[string]*relationTb.GroupSimpleUserID)
for _, groupID := range groupIDs {
userIDs := mapGroupUserIDs[groupID]
users := &relationTb.GroupSimpleUserID{}
if len(userIDs) > 0 {
utils.Sort(userIDs, true)
bi := big.NewInt(0)
bi.SetString(utils.Md5(strings.Join(userIDs, ";"))[0:8], 16)
users.Hash = bi.Uint64()
}
res[groupID] = users
}
return res, nil
return g.cache.GetGroupMemberHash1(ctx, groupIDs)
}
func (g *GroupDataBase) MapGroupMemberNum(ctx context.Context, groupIDs []string) (map[string]uint32, error) {

13
pkg/utils/utils_v2.go
View File

@ -412,3 +412,16 @@ func (o *sortSlice[E]) Swap(i, j int) {
type Ordered interface {
~int | ~int8 | ~int16 | ~int32 | ~int64 | ~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 | ~uintptr | ~float32 | ~float64 | ~string
}
func Unwrap(err error) error {
for err != nil {
unwrap, ok := err.(interface {
Unwrap() error
})
if !ok {
break
}
err = unwrap.Unwrap()
}
return err
}