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

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

Browse Source
This commit is contained in:
Gordon 2023-06-05 17:40:13 +08:00
commit 4b77839202
29 changed files with 1355 additions and 1150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config/config.yaml
View File

@ -240,7 +240,7 @@ iospush:
callback:
# callback url 需要自行更换callback url
callbackUrl: "http://127.0.0.1:8080/callback"
callbackUrl: "http://127.0.0.1:10008/callback"
# 开启关闭操作前后回调的配置
callbackBeforeSendSingleMsg:
enable: false # 回调是否启用

12
config/notification.yaml
View File

@ -12,7 +12,7 @@ groupCreated:
groupInfoSet:
isSendMsg: true
isSendMsg: false
reliabilityLevel: 1
unreadCount: false
offlinePush:
@ -23,7 +23,7 @@ groupInfoSet:
joinGroupApplication:
isSendMsg: true
isSendMsg: false
reliabilityLevel: 1
unreadCount: false
offlinePush:
@ -43,7 +43,7 @@ memberQuit:
ext: "memberQuit ext"
groupApplicationAccepted:
isSendMsg: true
isSendMsg: false
reliabilityLevel: 1
unreadCount: false
offlinePush:
@ -53,7 +53,7 @@ groupApplicationAccepted:
ext: "groupApplicationAccepted ext"
groupApplicationRejected:
isSendMsg: true
isSendMsg: false
reliabilityLevel: 1
unreadCount: false
offlinePush:
@ -157,7 +157,7 @@ groupMemberCancelMuted:
ext: "groupMemberCancelMuted ext"
groupMemberInfoSet:
isSendMsg: true
isSendMsg: false
reliabilityLevel: 1
unreadCount: false
offlinePush:
@ -259,7 +259,7 @@ friendInfoUpdated:
#####################user#########################
userInfoUpdated:
isSendMsg: true
isSendMsg: false
reliabilityLevel: 1
unreadCount: false
offlinePush:

2
internal/api/third.go
View File

@ -20,7 +20,7 @@ import (
func NewThird(discov discoveryregistry.SvcDiscoveryRegistry) *Third {
conn, err := discov.GetConn(context.Background(), config.Config.RpcRegisterName.OpenImThirdName)
if err != nil {
panic(err)
// panic(err)
}
return &Third{conn: conn, discov: discov}
}

12
internal/msggateway/callback.go
View File

@ -33,7 +33,7 @@ func CallbackUserOnline(ctx context.Context, userID string, platformID int, isAp
ConnID: connID,
}
resp := cbapi.CommonCallbackResp{}
return http.CallBackPostReturn(url(), &req, &resp, config.Config.Callback.CallbackUserOnline)
return http.CallBackPostReturn(ctx, url(), &req, &resp, config.Config.Callback.CallbackUserOnline)
}
func CallbackUserOffline(ctx context.Context, userID string, platformID int, connID string) error {
@ -54,7 +54,7 @@ func CallbackUserOffline(ctx context.Context, userID string, platformID int, con
ConnID: connID,
}
resp := &cbapi.CallbackUserOfflineResp{}
return http.CallBackPostReturn(url(), req, resp, config.Config.Callback.CallbackUserOffline)
return http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackUserOffline)
}
func CallbackUserKickOff(ctx context.Context, userID string, platformID int) error {
@ -74,7 +74,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
Seq: time.Now().UnixMilli(),
}
resp := &cbapi.CommonCallbackResp{}
return http.CallBackPostReturn(url(), req, resp, config.Config.Callback.CallbackUserOffline)
return http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackUserOffline)
}
//func callbackUserOnline(operationID, userID string, platformID int, token string, isAppBackground bool, connID string) cbApi.CommonCallbackResp {
@ -98,7 +98,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
// ConnID: connID,
// }
// callbackUserOnlineResp := &cbApi.CallbackUserOnlineResp{CommonCallbackResp: &callbackResp}
// if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackUserOnlineCommand, callbackUserOnlineReq, callbackUserOnlineResp, config.Config.Callback.CallbackUserOnline.CallbackTimeOut); err != nil {
// if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, constant.CallbackUserOnlineCommand, callbackUserOnlineReq, callbackUserOnlineResp, config.Config.Callback.CallbackUserOnline.CallbackTimeOut); err != nil {
// callbackResp.ErrCode = http2.StatusInternalServerError
// callbackResp.ErrMsg = err.Error()
// }
@ -123,7 +123,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
// ConnID: connID,
// }
// callbackUserOfflineResp := &cbApi.CallbackUserOfflineResp{CommonCallbackResp: &callbackResp}
// if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackUserOfflineCommand, callbackOfflineReq, callbackUserOfflineResp, config.Config.Callback.CallbackUserOffline.CallbackTimeOut); err != nil {
// if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, constant.CallbackUserOfflineCommand, callbackOfflineReq, callbackUserOfflineResp, config.Config.Callback.CallbackUserOffline.CallbackTimeOut); err != nil {
// callbackResp.ErrCode = http2.StatusInternalServerError
// callbackResp.ErrMsg = err.Error()
// }
@ -147,7 +147,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
// Seq: int(time.Now().UnixNano() / 1e6),
// }
// callbackUserKickOffResp := &cbApi.CallbackUserKickOffResp{CommonCallbackResp: &callbackResp}
// if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackUserKickOffCommand, callbackUserKickOffReq, callbackUserKickOffResp, config.Config.Callback.CallbackUserOffline.CallbackTimeOut); err != nil {
// if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, constant.CallbackUserKickOffCommand, callbackUserKickOffReq, callbackUserKickOffResp, config.Config.Callback.CallbackUserOffline.CallbackTimeOut); err != nil {
// callbackResp.ErrCode = http2.StatusInternalServerError
// callbackResp.ErrMsg = err.Error()
// }

6
internal/push/callback.go
View File

@ -39,7 +39,7 @@ func callbackOfflinePush(ctx context.Context, userIDs []string, msg *sdkws.MsgDa
Content: utils.GetContent(msg),
}
resp := &callbackstruct.CallbackBeforePushResp{}
err := http.CallBackPostReturn(url(), req, resp, config.Config.Callback.CallbackOfflinePush)
err := http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackOfflinePush)
if err != nil {
return err
}
@ -75,7 +75,7 @@ func callbackOnlinePush(ctx context.Context, userIDs []string, msg *sdkws.MsgDat
Content: utils.GetContent(msg),
}
resp := &callbackstruct.CallbackBeforePushResp{}
return http.CallBackPostReturn(url(), req, resp, config.Config.Callback.CallbackOnlinePush)
return http.CallBackPostReturn(ctx, url(), req, resp, config.Config.Callback.CallbackOnlinePush)
}
func callbackBeforeSuperGroupOnlinePush(ctx context.Context, groupID string, msg *sdkws.MsgData, pushToUserIDs *[]string) error {
@ -99,7 +99,7 @@ func callbackBeforeSuperGroupOnlinePush(ctx context.Context, groupID string, msg
Seq: msg.Seq,
}
resp := &callbackstruct.CallbackBeforeSuperGroupOnlinePushResp{}
if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, req, resp, config.Config.Callback.CallbackBeforeSuperGroupOnlinePush); err != nil {
if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, req, resp, config.Config.Callback.CallbackBeforeSuperGroupOnlinePush); err != nil {
return err
}
if len(resp.UserIDs) != 0 {

6
internal/push/offlinepush/getui/push.go
View File

@ -146,11 +146,11 @@ func (g *Client) request(ctx context.Context, url string, input interface{}, tok
header := map[string]string{"token": token}
resp := &Resp{}
resp.Data = output
return g.postReturn(config.Config.Push.Getui.PushUrl+url, header, input, resp, 3)
return g.postReturn(ctx, config.Config.Push.Getui.PushUrl+url, header, input, resp, 3)
}
func (g *Client) postReturn(url string, header map[string]string, input interface{}, output RespI, timeout int) error {
err := http2.PostReturn(url, header, input, output, timeout)
func (g *Client) postReturn(ctx context.Context, url string, header map[string]string, input interface{}, output RespI, timeout int) error {
err := http2.PostReturn(ctx, url, header, input, output, timeout)
if err != nil {
return err
}

6
internal/push/offlinepush/jpush/push.go
View File

@ -55,9 +55,9 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin
pushObj.SetMessage(&msg)
pushObj.SetOptions(&opt)
var resp interface{}
return j.request(pushObj, resp, 5)
return j.request(ctx, pushObj, resp, 5)
}
func (j *JPush) request(po body.PushObj, resp interface{}, timeout int) error {
return http2.PostReturn(config.Config.Push.Jpns.PushUrl, map[string]string{"Authorization": j.getAuthorization(config.Config.Push.Jpns.AppKey, config.Config.Push.Jpns.MasterSecret)}, po, resp, timeout)
func (j *JPush) request(ctx context.Context, po body.PushObj, resp interface{}, timeout int) error {
return http2.PostReturn(ctx, config.Config.Push.Jpns.PushUrl, map[string]string{"Authorization": j.getAuthorization(config.Config.Push.Jpns.AppKey, config.Config.Push.Jpns.MasterSecret)}, po, resp, timeout)
}

4
internal/push/push_handler.go
View File

@ -51,9 +51,13 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
err = c.pusher.Push2User(ctx, []string{pbData.MsgData.SendID, pbData.MsgData.RecvID}, pbData.MsgData)
}
if err != nil {
if err == errNoOfflinePusher {
log.ZWarn(ctx, "offline push failed", err, "msg", pbData.String())
} else {
log.ZError(ctx, "push failed", err, "msg", pbData.String())
}
}
}
func (ConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,

5
internal/push/push_rpc_server.go
View File

@ -8,6 +8,7 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/localcache"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry"
pbPush "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/push"
"google.golang.org/grpc"
@ -52,7 +53,11 @@ func (r *pushServer) PushMsg(ctx context.Context, pbData *pbPush.PushMsgReq) (re
err = r.pusher.Push2User(ctx, []string{pbData.MsgData.RecvID, pbData.MsgData.SendID}, pbData.MsgData)
}
if err != nil {
if err != errNoOfflinePusher {
return nil, err
} else {
log.ZWarn(ctx, "offline push failed", err, "msg", pbData.String())
}
}
return &pbPush.PushMsgResp{}, nil
}

5
internal/push/push_to_client.go
View File

@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"github.com/OpenIMSDK/Open-IM-Server/internal/push/offlinepush"
"github.com/OpenIMSDK/Open-IM-Server/internal/push/offlinepush/fcm"
"github.com/OpenIMSDK/Open-IM-Server/internal/push/offlinepush/getui"
@ -35,6 +36,8 @@ type Pusher struct {
successCount int
}
var errNoOfflinePusher = errors.New("no offlinePusher is configured")
func NewPusher(client discoveryregistry.SvcDiscoveryRegistry, offlinePusher offlinepush.OfflinePusher, database controller.PushDatabase,
groupLocalCache *localcache.GroupLocalCache, conversationLocalCache *localcache.ConversationLocalCache) *Pusher {
rpcclient.NewGroupClient(client)
@ -286,7 +289,7 @@ func (p *Pusher) GetOfflinePushOpts(msg *sdkws.MsgData) (opts *offlinepush.Opts,
func (p *Pusher) getOfflinePushInfos(conversationID string, msg *sdkws.MsgData) (title, content string, opts *offlinepush.Opts, err error) {
if p.offlinePusher == nil {
err = errors.New("no offlinePusher is configured")
err = errNoOfflinePusher
return
}
type AtContent struct {

5
internal/rpc/friend/callback.go
View File

@ -6,11 +6,13 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/http"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
pbfriend "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/friend"
)
func CallbackBeforeAddFriend(ctx context.Context, req *pbfriend.ApplyToAddFriendReq) error {
log.ZInfo(ctx, "CallbackBeforeAddFriend", "in", req)
if !config.Config.Callback.CallbackBeforeAddFriend.Enable {
return nil
}
@ -22,5 +24,6 @@ func CallbackBeforeAddFriend(ctx context.Context, req *pbfriend.ApplyToAddFriend
OperationID: mcontext.GetOperationID(ctx),
}
resp := &cbapi.CallbackBeforeAddFriendResp{}
return http.CallBackPostReturn(config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddFriend)
defer log.ZInfo(ctx, "CallbackBeforeAddFriend", "out", &resp)
return http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeAddFriend)
}

6
internal/rpc/friend/friend.go
View File

@ -63,12 +63,12 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbfriend.Apply
if err := tokenverify.CheckAccessV3(ctx, req.FromUserID); err != nil {
return nil, err
}
if err := CallbackBeforeAddFriend(ctx, req); err != nil && err != errs.ErrCallbackContinue {
return nil, err
}
if req.ToUserID == req.FromUserID {
return nil, errs.ErrCanNotAddYourself.Wrap()
}
if err := CallbackBeforeAddFriend(ctx, req); err != nil && err != errs.ErrCallbackContinue {
return nil, err
}
if _, err := s.userRpcClient.GetUsersInfoMap(ctx, []string{req.ToUserID, req.FromUserID}); err != nil {
return nil, err
}

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

@ -42,7 +42,7 @@ func CallbackBeforeCreateGroup(ctx context.Context, req *group.CreateGroupReq) (
})
}
resp := &callbackstruct.CallbackBeforeCreateGroupResp{}
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeCreateGroup)
err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeCreateGroup)
if err != nil {
return err
}
@ -74,7 +74,7 @@ func CallbackBeforeMemberJoinGroup(ctx context.Context, groupMember *relation.Gr
GroupEx: groupEx,
}
resp := &callbackstruct.CallbackBeforeMemberJoinGroupResp{}
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeMemberJoinGroup)
err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeMemberJoinGroup)
if err != nil {
return err
}
@ -111,7 +111,7 @@ func CallbackBeforeSetGroupMemberInfo(ctx context.Context, req *group.SetGroupMe
callbackReq.Ex = &req.Ex.Value
}
resp := &callbackstruct.CallbackBeforeSetGroupMemberInfoResp{}
err = http.CallBackPostReturn(config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo)
err = http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, callbackReq, resp, config.Config.Callback.CallbackBeforeSetGroupMemberInfo)
if err != nil {
return err
}

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

@ -485,34 +485,30 @@ func (s *groupServer) KickGroupMember(ctx context.Context, req *pbGroup.KickGrou
}
memberMap := make(map[string]*relationTb.GroupMemberModel)
for i, member := range members {
if member.RoleLevel == constant.GroupOwner {
return nil, errs.ErrArgs.Wrap("can not kick group owner")
}
memberMap[member.UserID] = members[i]
}
isAppManagerUid := tokenverify.IsAppManagerUid(ctx)
opMember := memberMap[opUserID]
for _, userID := range req.KickedUserIDs {
if _, ok := memberMap[userID]; !ok {
member, ok := memberMap[userID]
if !ok {
return nil, errs.ErrUserIDNotFound.Wrap(userID)
}
if !isAppManagerUid {
if opMember == nil {
return nil, errs.ErrNoPermission.Wrap("opUserID no in group")
}
if !tokenverify.IsAppManagerUid(ctx) {
member := memberMap[opUserID]
if member == nil {
return nil, errs.ErrNoPermission.Wrap(fmt.Sprintf("opUserID %s no in group", opUserID))
}
switch member.RoleLevel {
switch opMember.RoleLevel {
case constant.GroupOwner:
case constant.GroupAdmin:
for _, member := range members {
if member.UserID == opUserID {
continue
}
if member.RoleLevel == constant.GroupOwner || member.RoleLevel == constant.GroupAdmin {
return nil, errs.ErrNoPermission.Wrap("userID:" + member.UserID)
}
return nil, errs.ErrNoPermission.Wrap("group admins cannot remove the group owner and other admins")
}
case constant.GroupOrdinaryUsers:
return nil, errs.ErrNoPermission.Wrap("opUserID no permission")
default:
return nil, errs.ErrNoPermission.Wrap("opUserID is OrdinaryUser")
return nil, errs.ErrNoPermission.Wrap("opUserID roleLevel unknown")
}
}
}
userIDs, err := s.GroupDatabase.FindGroupMemberUserID(ctx, req.GroupID)
@ -1343,8 +1339,18 @@ func (s *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbGroup.SetGr
return nil, err
}
for _, member := range req.Members {
if member.RoleLevel != nil {
switch member.RoleLevel.Value {
case constant.GroupAdmin:
s.Notification.GroupMemberSetToAdminNotification(ctx, member.GroupID, member.UserID)
case constant.GroupOrdinaryUsers:
s.Notification.GroupMemberSetToOrdinaryUserNotification(ctx, member.GroupID, member.UserID)
}
}
if member.Nickname != nil || member.FaceURL != nil || member.Ex != nil {
s.Notification.GroupMemberInfoSetNotification(ctx, member.GroupID, member.UserID)
}
}
return resp, nil
}

32
internal/rpc/msg/as_read.go
View File

@ -3,9 +3,41 @@ package msg
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
)
func (m *msgServer) MarkMsgsAsRead(ctx context.Context, req *msg.MarkMsgsAsReadReq) (resp *msg.MarkMsgsAsReadResp, err error) {
conversations, err := m.Conversation.GetConversationsByConversationID(ctx, []string{req.ConversationID})
if err != nil {
return
}
var recvID string
if conversations[0].ConversationType == constant.SingleChatType || conversations[0].ConversationType == constant.NotificationChatType {
if req.UserID == conversations[0].OwnerUserID {
recvID = conversations[0].UserID
} else {
recvID = conversations[0].OwnerUserID
}
} else if conversations[0].ConversationType == constant.SuperGroupChatType {
recvID = conversations[0].GroupID
}
err = m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.ConversationID, req.UserID, req.Seqs)
if err != nil {
return
}
if err = m.sendMarkAsReadNotification(ctx, req.ConversationID, req.UserID, recvID, req.Seqs); err != nil {
return
}
return &msg.MarkMsgsAsReadResp{}, nil
}
func (m *msgServer) sendMarkAsReadNotification(ctx context.Context, conversationID string, sendID, recvID string, seqs []int64) error {
// tips := &sdkws.MarkAsReadTips{
// MarkAsReadUserID: sendID,
// ConversationID: conversationID,
// Seqs: seqs,
// }
// m.notificationSender.NotificationWithSesstionType(ctx)
return nil
}

10
internal/rpc/msg/callback.go
View File

@ -47,7 +47,7 @@ func callbackBeforeSendSingleMsg(ctx context.Context, msg *pbChat.SendMsgReq) er
RecvID: msg.MsgData.RecvID,
}
resp := &cbapi.CallbackBeforeSendSingleMsgResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackBeforeSendSingleMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackBeforeSendSingleMsg)
}
func callbackAfterSendSingleMsg(ctx context.Context, msg *pbChat.SendMsgReq) error {
@ -59,7 +59,7 @@ func callbackAfterSendSingleMsg(ctx context.Context, msg *pbChat.SendMsgReq) err
RecvID: msg.MsgData.RecvID,
}
resp := &cbapi.CallbackAfterSendSingleMsgResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendSingleMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendSingleMsg)
}
func callbackBeforeSendGroupMsg(ctx context.Context, msg *pbChat.SendMsgReq) error {
@ -71,7 +71,7 @@ func callbackBeforeSendGroupMsg(ctx context.Context, msg *pbChat.SendMsgReq) err
GroupID: msg.MsgData.GroupID,
}
resp := &cbapi.CallbackBeforeSendGroupMsgResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendSingleMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendSingleMsg)
}
func callbackAfterSendGroupMsg(ctx context.Context, msg *pbChat.SendMsgReq) error {
@ -83,7 +83,7 @@ func callbackAfterSendGroupMsg(ctx context.Context, msg *pbChat.SendMsgReq) erro
GroupID: msg.MsgData.GroupID,
}
resp := &cbapi.CallbackAfterSendGroupMsgResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
}
func callbackMsgModify(ctx context.Context, msg *pbChat.SendMsgReq) error {
@ -94,7 +94,7 @@ func callbackMsgModify(ctx context.Context, msg *pbChat.SendMsgReq) error {
CommonCallbackReq: toCommonCallback(ctx, msg, constant.CallbackMsgModifyCommand),
}
resp := &cbapi.CallbackMsgModifyCommandResp{}
if err := http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg); err != nil {
if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg); err != nil {
return err
}
if resp.Content != nil {

10
internal/rpc/msg/extend_msg_callback.go
View File

@ -28,14 +28,14 @@ func callbackSetMessageReactionExtensions(ctx context.Context, setReq *msg.SetMe
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbapi.CallbackBeforeSetMessageReactionExtResp{}
if err := http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg); err != nil {
if err := http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg); err != nil {
return err
}
setReq.MsgFirstModifyTime = resp.MsgFirstModifyTime
return nil
}
func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessagesReactionExtensionsReq) error {
func callbackDeleteMessageReactionExtensions(ctx context.Context, setReq *msg.DeleteMessagesReactionExtensionsReq) error {
if !config.Config.Callback.CallbackAfterSendGroupMsg.Enable {
return nil
}
@ -51,7 +51,7 @@ func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessagesReactionE
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbapi.CallbackDeleteMessageReactionExtResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
}
func callbackGetMessageListReactionExtensions(ctx context.Context, getReq *msg.GetMessagesReactionExtensionsReq) error {
@ -67,7 +67,7 @@ func callbackGetMessageListReactionExtensions(ctx context.Context, getReq *msg.G
TypeKeyList: getReq.TypeKeys,
}
resp := &cbapi.CallbackGetMessageListReactionExtResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
}
func callbackAddMessageReactionExtensions(ctx context.Context, setReq *msg.ModifyMessageReactionExtensionsReq) error {
@ -84,5 +84,5 @@ func callbackAddMessageReactionExtensions(ctx context.Context, setReq *msg.Modif
MsgFirstModifyTime: setReq.MsgFirstModifyTime,
}
resp := &cbapi.CallbackAddMessageReactionExtResp{}
return http.CallBackPostReturn(cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
return http.CallBackPostReturn(ctx, cbURL(), req, resp, config.Config.Callback.CallbackAfterSendGroupMsg)
}

4
pkg/common/constant/constant.go
View File

@ -93,7 +93,7 @@ const (
ClearConversationNotification = 2101
DeleteMsgsNotification = 2102
NotificationEnd = 3000
NotificationEnd = 5000
//status
MsgNormal = 1
@ -334,3 +334,5 @@ const (
)
const OpenIMCommonConfigKey = "OpenIMServerConfig"
const CallbackCommand = "command"

2
pkg/common/db/cache/black.go vendored
View File

@ -62,6 +62,6 @@ func (b *BlackCacheRedis) GetBlackIDs(ctx context.Context, userID string) (black
func (b *BlackCacheRedis) DelBlackIDs(ctx context.Context, userID string) BlackCache {
cache := b.NewCache()
cache.AddKeys(userID)
cache.AddKeys(b.getBlackIDsKey(userID))
return cache
}

2
pkg/common/db/controller/black.go
View File

@ -5,6 +5,7 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
)
@ -68,6 +69,7 @@ func (b *blackDatabase) CheckIn(ctx context.Context, userID1, userID2 string) (i
if err != nil {
return
}
log.ZDebug(ctx, "blackIDs", "user1BlackIDs", userID1BlackIDs, "user2BlackIDs", userID2BlackIDs)
return utils.IsContain(userID2, userID1BlackIDs), utils.IsContain(userID1, userID2BlackIDs), nil
}

6
pkg/common/db/controller/msg.go
View File

@ -34,6 +34,8 @@ type CommonMsgDatabase interface {
BatchInsertChat2DB(ctx context.Context, conversationID string, msgs []*sdkws.MsgData, currentMaxSeq int64) error
// 撤回消息
RevokeMsg(ctx context.Context, conversationID string, seq int64, revoke *unRelationTb.RevokeModel) error
// mark as read
MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, seqs []int64) error
// 刪除redis中消息缓存
DeleteMessageFromCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) error
// incrSeq然后批量插入缓存
@ -297,6 +299,10 @@ func (db *commonMsgDatabase) RevokeMsg(ctx context.Context, conversationID strin
return db.BatchInsertBlock(ctx, conversationID, []any{revoke}, updateKeyRevoke, seq)
}
func (db *commonMsgDatabase) MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, seqs []int64) error {
return db.msgDocDatabase.MarkSingleChatMsgsAsRead(ctx, userID, conversationID, seqs)
}
func (db *commonMsgDatabase) DeleteMessageFromCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) error {
return db.cache.DeleteMessageFromCache(ctx, conversationID, msgs)
}

16
pkg/common/db/ormutil/utils.go
View File

@ -1,8 +1,10 @@
package ormutil
import (
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
"gorm.io/gorm"
"strings"
)
func GormPage[E any](db *gorm.DB, pageNumber, showNumber int32) (uint32, []*E, error) {
@ -20,16 +22,14 @@ func GormPage[E any](db *gorm.DB, pageNumber, showNumber int32) (uint32, []*E, e
func GormSearch[E any](db *gorm.DB, fields []string, value string, pageNumber, showNumber int32) (uint32, []*E, error) {
if len(fields) > 0 && value != "" {
value = "%" + value + "%"
if len(fields) == 1 {
db = db.Where(fields[0]+" like ?", value)
} else {
t := db
val := "%" + value + "%"
arr := make([]string, 0, len(fields))
vals := make([]interface{}, 0, len(fields))
for _, field := range fields {
t = t.Or(field+" like ?", value)
}
db = db.Where(t)
arr = append(arr, fmt.Sprintf("`%s` like ?", field))
vals = append(vals, val)
}
db = db.Where(strings.Join(arr, " or "), vals...)
}
return GormPage[E](db, pageNumber, showNumber)
}

5
pkg/common/db/relation/black_model.go
View File

@ -2,6 +2,7 @@ package relation
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/ormutil"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
@ -51,11 +52,11 @@ func (b *BlackGorm) FindOwnerBlacks(ctx context.Context, ownerUserID string, pag
if err != nil {
return nil, 0, utils.Wrap(err, "")
}
totalUint32, blacks, err := ormutil.GormPage[relation.BlackModel](b.db(ctx), pageNumber, showNumber)
totalUint32, blacks, err := ormutil.GormPage[relation.BlackModel](b.db(ctx).Where("owner_user_id = ?", ownerUserID), pageNumber, showNumber)
total = int64(totalUint32)
return
}
func (b *BlackGorm) FindBlackUserIDs(ctx context.Context, ownerUserID string) (blackUserIDs []string, err error) {
return blackUserIDs, utils.Wrap(b.db(ctx).Where("owner_user_id = ?", blackUserIDs).Pluck("block_user_id", &blackUserIDs).Error, "")
return blackUserIDs, utils.Wrap(b.db(ctx).Where("owner_user_id = ?", ownerUserID).Pluck("block_user_id", &blackUserIDs).Error, "")
}

1
pkg/common/db/table/unrelation/msg.go
View File

@ -78,6 +78,7 @@ type MsgDocModelInterface interface {
DeleteDocs(ctx context.Context, docIDs []string) error
GetMsgDocModelByIndex(ctx context.Context, conversationID string, index, sort int64) (*MsgDocModel, error)
DeleteMsgsInOneDocByIndex(ctx context.Context, docID string, indexes []int) error
MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, seqs []int64) error
}
func (MsgDocModel) TableName() string {

14
pkg/common/db/unrelation/msg.go
View File

@ -5,9 +5,10 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"strings"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
table "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
@ -263,7 +264,7 @@ func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, userID stri
Detail: string(tipsData),
}
content, _ := json.Marshal(&elem)
msg.Msg.ContentType = constant.Revoke
msg.Msg.ContentType = constant.MsgRevokeNotification
msg.Msg.Content = string(content)
}
msgs = append(msgs, msg)
@ -278,3 +279,12 @@ func (m *MsgMongoDriver) IsExistDocID(ctx context.Context, docID string) (bool,
}
return count > 0, nil
}
func (m *MsgMongoDriver) MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, seqs []int64) error {
indexs := make([]int64, 0, len(seqs))
for _, seq := range seqs {
indexs = append(indexs, m.model.GetMsgIndex(seq))
}
return nil
}

36
pkg/common/http/http_client.go
View File

@ -8,16 +8,21 @@ package http
import (
"bytes"
"context"
"encoding/json"
"github.com/OpenIMSDK/Open-IM-Server/pkg/callbackstruct"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
"io"
"io/ioutil"
"net/http"
urlLib "net/url"
"time"
)
var client http.Client
func Get(url string) (response []byte, err error) {
client := http.Client{Timeout: 5 * time.Second}
resp, err := client.Get(url)
@ -32,36 +37,41 @@ func Get(url string) (response []byte, err error) {
return body, nil
}
// application/json; charset=utf-8
func Post(url string, header map[string]string, data interface{}, timeout int) (content []byte, err error) {
func Post(ctx context.Context, url string, header map[string]string, data interface{}, timeout int) (content []byte, err error) {
if timeout > 0 {
var cancel func()
ctx, cancel = context.WithTimeout(ctx, time.Second*time.Duration(timeout))
defer cancel()
}
jsonStr, err := json.Marshal(data)
if err != nil {
return nil, err
}
req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonStr))
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonStr))
if err != nil {
return nil, err
}
if operationID, _ := ctx.Value(constant.OperationID).(string); operationID != "" {
req.Header.Set(constant.OperationID, operationID)
}
for k, v := range header {
req.Header.Set(k, v)
}
req.Close = true
req.Header.Add("content-type", "application/json; charset=utf-8")
client := &http.Client{Timeout: time.Duration(timeout) * time.Second}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
result, err := ioutil.ReadAll(resp.Body)
result, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
return result, nil
}
func PostReturn(url string, header map[string]string, input, output interface{}, timeOutSecond int) error {
b, err := Post(url, header, input, timeOutSecond)
func PostReturn(ctx context.Context, url string, header map[string]string, input, output interface{}, timeOutSecond int) error {
b, err := Post(ctx, url, header, input, timeOutSecond)
if err != nil {
return err
}
@ -69,11 +79,11 @@ func PostReturn(url string, header map[string]string, input, output interface{},
return err
}
func callBackPostReturn(url, command string, input interface{}, output callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error {
func callBackPostReturn(ctx context.Context, url, command string, input interface{}, output callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error {
v := urlLib.Values{}
v.Set("callbackCommand", command)
v.Set(constant.CallbackCommand, command)
url = url + "?" + v.Encode()
b, err := Post(url, nil, input, callbackConfig.CallbackTimeOut)
b, err := Post(ctx, url, nil, input, callbackConfig.CallbackTimeOut)
if err != nil {
if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue {
return errs.ErrCallbackContinue
@ -89,6 +99,6 @@ func callBackPostReturn(url, command string, input interface{}, output callbacks
return output.Parse()
}
func CallBackPostReturn(url string, req callbackstruct.CallbackReq, resp callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error {
return callBackPostReturn(url, req.GetCallbackCommand(), req, resp, callbackConfig)
func CallBackPostReturn(ctx context.Context, url string, req callbackstruct.CallbackReq, resp callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error {
return callBackPostReturn(ctx, url, req.GetCallbackCommand(), req, resp, callbackConfig)
}

2188
pkg/proto/sdkws/sdkws.pb.go
View File

File diff suppressed because it is too large Load Diff

17
pkg/proto/sdkws/sdkws.proto
View File

@ -181,11 +181,12 @@ message MsgData {
int64 sendTime = 15;
int64 createTime = 16;
int32 status = 17;
map<string, bool> options = 18;
OfflinePushInfo offlinePushInfo = 19;
repeated string atUserIDList = 20;
string attachedInfo = 21;
string ex = 22;
bool isRead = 18;
map<string, bool> options = 19;
OfflinePushInfo offlinePushInfo = 20;
repeated string atUserIDList = 21;
string attachedInfo = 22;
string ex = 23;
}
message PushMessages{
map<string, PullMsgs> msgs = 1;
@ -453,6 +454,12 @@ message DeleteMsgsTips {
repeated int64 seqs = 3;
}
message MarkAsReadTips {
string markAsReadUserID = 1;
string conversationID = 2;
repeated int64 seqs = 3;
}
///////////////////signal//////////////
message SignalReq {
oneof payload {

25
pkg/rpcclient/notification/group.go
View File

@ -673,10 +673,10 @@ func (g *GroupNotificationSender) GroupMemberInfoSetNotification(ctx context.Con
return err
}
tips := &sdkws.GroupMemberInfoSetTips{Group: group, OpUser: user[mcontext.GetOpUserID(ctx)], ChangedUser: user[groupMemberUserID]}
return g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberCancelMutedNotification, tips)
return g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberInfoSetNotification, tips)
}
func (g *GroupNotificationSender) GroupMemberSetToAdminNotification(ctx context.Context, groupID, groupMemberUserID string, notificationType int32) (err error) {
func (g *GroupNotificationSender) GroupMemberSetToAdminNotification(ctx context.Context, groupID, groupMemberUserID string) (err error) {
defer log.ZDebug(ctx, "return")
defer func() {
if err != nil {
@ -692,7 +692,26 @@ func (g *GroupNotificationSender) GroupMemberSetToAdminNotification(ctx context.
return err
}
tips := &sdkws.GroupMemberInfoSetTips{Group: group, OpUser: user[mcontext.GetOpUserID(ctx)], ChangedUser: user[groupMemberUserID]}
return g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberCancelMutedNotification, tips)
return g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToAdminNotification, tips)
}
func (g *GroupNotificationSender) GroupMemberSetToOrdinaryUserNotification(ctx context.Context, groupID, groupMemberUserID string) (err error) {
defer log.ZDebug(ctx, "return")
defer func() {
if err != nil {
log.ZError(ctx, utils.GetFuncName(1)+" failed", err)
}
}()
group, err := g.getGroupInfo(ctx, groupID)
if err != nil {
return err
}
user, err := g.getGroupMemberMap(ctx, groupID, []string{mcontext.GetOpUserID(ctx), groupMemberUserID})
if err != nil {
return err
}
tips := &sdkws.GroupMemberInfoSetTips{Group: group, OpUser: user[mcontext.GetOpUserID(ctx)], ChangedUser: user[groupMemberUserID]}
return g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToOrdinaryUserNotification, tips)
}
func (g *GroupNotificationSender) MemberEnterDirectlyNotification(ctx context.Context, groupID string, entrantUserID string) (err error) {