msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Merge pull request #3205 from openimsdk/cherry-pick-964ee7a

Browse Source 
feat: sending messages supports returning fields modified by webhook [Created by @withchao from #3192]
pull/3220/head
chao 7 months ago committed by GitHub
parent 05da0eec30 7e9bb7bc76
commit 1e54235263
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 208 additions and 33 deletions
Show Stats Download Patch File Download Diff File

2
go.mod
Unescape View File

@ -12,7 +12,7 @@ require (
github.com/gorilla/websocket v1.5.1
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/protocol v0.0.72-alpha.75
github.com/openimsdk/protocol v0.0.72-alpha.79
github.com/openimsdk/tools v0.0.50-alpha.74
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0

4
go.sum
Unescape View File

@ -347,8 +347,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM=
github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/protocol v0.0.72-alpha.75 h1:WlmBn8g2Fvv21g8TEFVbolmbw2rU0sN9kj6sQaDK7cA=
github.com/openimsdk/protocol v0.0.72-alpha.75/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/protocol v0.0.72-alpha.79 h1:e46no8WVAsmTzyy405klrdoUiG7u+1ohDsXvQuFng4s=
github.com/openimsdk/protocol v0.0.72-alpha.79/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/tools v0.0.50-alpha.74 h1:yh10SiMiivMEjicEQg+QAsH4pvaO+4noMPdlw+ew0Kc=
github.com/openimsdk/tools v0.0.50-alpha.74/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=

85
internal/api/msg.go
Unescape View File

@ -17,10 +17,12 @@ package api
import (
"encoding/base64"
"encoding/json"
"sync"
"github.com/gin-gonic/gin"
"github.com/go-playground/validator/v10"
"github.com/mitchellh/mapstructure"
"google.golang.org/protobuf/reflect/protoreflect"
"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
@ -41,6 +43,39 @@ import (
"github.com/openimsdk/tools/utils/timeutil"
)
var (
msgDataDescriptor []protoreflect.FieldDescriptor
msgDataDescriptorOnce sync.Once
)
func getMsgDataDescriptor() []protoreflect.FieldDescriptor {
msgDataDescriptorOnce.Do(func() {
skip := make(map[string]struct{})
respFields := new(msg.SendMsgResp).ProtoReflect().Descriptor().Fields()
for i := 0; i < respFields.Len(); i++ {
field := respFields.Get(i)
if !field.HasJSONName() {
continue
}
skip[field.JSONName()] = struct{}{}
}
fields := new(sdkws.MsgData).ProtoReflect().Descriptor().Fields()
num := fields.Len()
msgDataDescriptor = make([]protoreflect.FieldDescriptor, 0, num)
for i := 0; i < num; i++ {
field := fields.Get(i)
if !field.HasJSONName() {
continue
}
if _, ok := skip[field.JSONName()]; ok {
continue
}
msgDataDescriptor = append(msgDataDescriptor, fields.Get(i))
}
})
return msgDataDescriptor
}
type MessageApi struct {
Client msg.MsgClient
userClient *rpcli.UserClient
@ -197,6 +232,42 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
return m.newUserSendMsgReq(c, &req), nil
}
func (m *MessageApi) getModifyFields(req, respModify *sdkws.MsgData) map[string]any {
if req == nil || respModify == nil {
return nil
}
fields := make(map[string]any)
reqProtoReflect := req.ProtoReflect()
respProtoReflect := respModify.ProtoReflect()
for _, descriptor := range getMsgDataDescriptor() {
reqValue := reqProtoReflect.Get(descriptor)
respValue := respProtoReflect.Get(descriptor)
if !reqValue.Equal(respValue) {
val := respValue.Interface()
name := descriptor.JSONName()
if name == "content" {
if bs, ok := val.([]byte); ok {
val = string(bs)
}
}
fields[name] = val
}
}
if len(fields) == 0 {
fields = nil
}
return fields
}
func (m *MessageApi) ginRespSendMsg(c *gin.Context, req *msg.SendMsgReq, resp *msg.SendMsgResp) {
res := m.getModifyFields(req.MsgData, resp.Modify)
resp.Modify = nil
apiresp.GinSuccess(c, &apistruct.SendMsgResp{
SendMsgResp: resp,
Modify: res,
})
}
// SendMessage handles the sending of a message. It's an HTTP handler function to be used with Gin framework.
func (m *MessageApi) SendMessage(c *gin.Context) {
// Initialize a request struct for sending a message.
@ -250,7 +321,7 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
}
// Respond with a success message and the response payload.
apiresp.GinSuccess(c, respPb)
m.ginRespSendMsg(c, sendMsgReq, respPb)
}
func (m *MessageApi) SendBusinessNotification(c *gin.Context) {
@ -316,7 +387,7 @@ func (m *MessageApi) SendBusinessNotification(c *gin.Context) {
apiresp.GinError(c, err)
return
}
apiresp.GinSuccess(c, respPb)
m.ginRespSendMsg(c, &sendMsgReq, respPb)
}
func (m *MessageApi) BatchSendMsg(c *gin.Context) {
@ -370,6 +441,7 @@ func (m *MessageApi) BatchSendMsg(c *gin.Context) {
ClientMsgID: rpcResp.ClientMsgID,
SendTime: rpcResp.SendTime,
RecvID: recvID,
Modify: m.getModifyFields(sendMsgReq.MsgData, rpcResp.Modify),
})
}
apiresp.GinSuccess(c, resp)
@ -432,7 +504,11 @@ func (m *MessageApi) SendSimpleMessage(c *gin.Context) {
Ex: req.Ex,
}
respPb, err := m.Client.SendMsg(c, &msg.SendMsgReq{MsgData: msgData})
sendReq := &msg.SendMsgReq{
MsgData: msgData,
}
respPb, err := m.Client.SendMsg(c, sendReq)
if err != nil {
apiresp.GinError(c, err)
return
@ -448,8 +524,7 @@ func (m *MessageApi) SendSimpleMessage(c *gin.Context) {
apiresp.GinError(c, err)
return
}
apiresp.GinSuccess(c, respPb)
m.ginRespSendMsg(c, sendReq, respPb)
}
func (m *MessageApi) CheckMsgIsSendSuccess(c *gin.Context) {

17
internal/rpc/msg/callback.go
Unescape View File

@ -17,9 +17,11 @@ package msg
import (
"context"
"encoding/base64"
"encoding/json"
"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
"github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/utils/stringutil"
cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
@ -136,11 +138,11 @@ func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.
m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
}
func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error {
func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq, beforeMsgData **sdkws.MsgData) error {
return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
if msg.MsgData.ContentType != constant.Text {
return nil
}
//if msg.MsgData.ContentType != constant.Text {
// return nil
//}
if !filterBeforeMsg(msg, before) {
return nil
}
@ -151,9 +153,14 @@ func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.B
if err := m.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
return err
}
if beforeMsgData != nil {
*beforeMsgData = proto.Clone(msg.MsgData).(*sdkws.MsgData)
}
if resp.Content != nil {
msg.MsgData.Content = []byte(*resp.Content)
if err := json.Unmarshal(msg.MsgData.Content, &struct{}{}); err != nil {
return errs.ErrArgs.WrapMsg("webhook msg modify content is not json", "content", string(msg.MsgData.Content))
}
}
datautil.NotNilReplace(msg.MsgData.OfflinePushInfo, resp.OfflinePushInfo)
datautil.NotNilReplace(&msg.MsgData.RecvID, resp.RecvID)

43
internal/rpc/msg/send.go
Unescape View File

@ -29,26 +29,44 @@ import (
"github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils/datautil"
"google.golang.org/protobuf/proto"
)
func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
if req.MsgData != nil {
if req.MsgData == nil {
return nil, errs.ErrArgs.WrapMsg("msgData is nil")
}
before := new(*sdkws.MsgData)
resp, err := m.sendMsg(ctx, req, before)
if err != nil {
return nil, err
}
if *before != nil && proto.Equal(*before, req.MsgData) == false {
resp.Modify = req.MsgData
}
return resp, nil
}
func (m *msgServer) sendMsg(ctx context.Context, req *pbmsg.SendMsgReq, before **sdkws.MsgData) (*pbmsg.SendMsgResp, error) {
m.encapsulateMsgData(req.MsgData)
if req.MsgData.ContentType == constant.Stream {
if err := m.handlerStreamMsg(ctx, req.MsgData); err != nil {
return nil, err
}
}
switch req.MsgData.SessionType {
case constant.SingleChatType:
return m.sendMsgSingleChat(ctx, req)
return m.sendMsgSingleChat(ctx, req, before)
case constant.NotificationChatType:
return m.sendMsgNotification(ctx, req)
return m.sendMsgNotification(ctx, req, before)
case constant.ReadGroupChatType:
return m.sendMsgGroupChat(ctx, req)
return m.sendMsgGroupChat(ctx, req, before)
default:
return nil, errs.ErrArgs.WrapMsg("unknown sessionType")
}
}
return nil, errs.ErrArgs.WrapMsg("msgData is nil")
}
func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq, before **sdkws.MsgData) (resp *pbmsg.SendMsgResp, err error) {
if err = m.messageVerification(ctx, req); err != nil {
prommetrics.GroupChatMsgProcessFailedCounter.Inc()
return nil, err
@ -57,7 +75,7 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq)
if err = m.webhookBeforeSendGroupMsg(ctx, &m.config.WebhooksConfig.BeforeSendGroupMsg, req); err != nil {
return nil, err
}
if err := m.webhookBeforeMsgModify(ctx, &m.config.WebhooksConfig.BeforeMsgModify, req); err != nil {
if err := m.webhookBeforeMsgModify(ctx, &m.config.WebhooksConfig.BeforeMsgModify, req, before); err != nil {
return nil, err
}
err = m.MsgDatabase.MsgToMQ(ctx, conversationutil.GenConversationUniqueKeyForGroup(req.MsgData.GroupID), req.MsgData)
@ -139,7 +157,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
}
}
func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbmsg.SendMsgReq, before **sdkws.MsgData) (resp *pbmsg.SendMsgResp, err error) {
if err := m.MsgDatabase.MsgToMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
return nil, err
}
@ -151,7 +169,7 @@ func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbmsg.SendMsgR
return resp, nil
}
func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq, before **sdkws.MsgData) (resp *pbmsg.SendMsgResp, err error) {
if err := m.messageVerification(ctx, req); err != nil {
return nil, err
}
@ -171,12 +189,11 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
}
if !isSend {
prommetrics.SingleChatMsgProcessFailedCounter.Inc()
return nil, nil
return nil, errs.ErrArgs.WrapMsg("message is not sent")
} else {
if err := m.webhookBeforeMsgModify(ctx, &m.config.WebhooksConfig.BeforeMsgModify, req); err != nil {
if err := m.webhookBeforeMsgModify(ctx, &m.config.WebhooksConfig.BeforeMsgModify, req, before); err != nil {
return nil, err
}
if err := m.MsgDatabase.MsgToMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
prommetrics.SingleChatMsgProcessFailedCounter.Inc()
return nil, err

12
pkg/apistruct/manage.go
Unescape View File

@ -15,6 +15,7 @@
package apistruct
import (
pbmsg "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws"
)
@ -139,4 +140,15 @@ type SingleReturnResult struct {
// RecvID uniquely identifies the receiver of the message.
RecvID string `json:"recvID"`
// Modify fields modified via webhook.
Modify map[string]any `json:"modify,omitempty"`
}
type SendMsgResp struct {
// SendMsgResp original response.
*pbmsg.SendMsgResp
// Modify fields modified via webhook.
Modify map[string]any `json:"modify,omitempty"`
}

1
pkg/apistruct/msg_test.go
Unescape View File

@ -1 +0,0 @@
package apistruct

65
test/webhook/msgmodify/main.go
Unescape View File

@ -0,0 +1,65 @@
package main
import (
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"github.com/gin-gonic/gin"
cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
"github.com/openimsdk/protocol/constant"
)
func main() {
g := gin.Default()
g.POST("/callbackExample/callbackBeforeMsgModifyCommand", toGin(handlerMsg))
if err := g.Run(":10006"); err != nil {
panic(err)
}
}
func toGin[R any](fn func(c *gin.Context, req *R)) gin.HandlerFunc {
return func(c *gin.Context) {
body, err := io.ReadAll(c.Request.Body)
if err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
fmt.Printf("HTTP %s %s %s\n", c.Request.Method, c.Request.URL, body)
var req R
if err := json.Unmarshal(body, &req); err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
fn(c, &req)
}
}
func handlerMsg(c *gin.Context, req *cbapi.CallbackMsgModifyCommandReq) {
var resp cbapi.CallbackMsgModifyCommandResp
if req.ContentType != constant.Text {
c.JSON(http.StatusOK, &resp)
return
}
var textElem struct {
Content string `json:"content"`
}
if err := json.Unmarshal([]byte(req.Content), &textElem); err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
const word = "xxx"
if strings.Contains(textElem.Content, word) {
textElem.Content = strings.ReplaceAll(textElem.Content, word, strings.Repeat("*", len(word)))
content, err := json.Marshal(&textElem)
if err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
tmp := string(content)
resp.Content = &tmp
}
c.JSON(http.StatusOK, &resp)
}
Write Preview
Loading…
Cancel
Save