From 49ec272f88437daf1029e4e06f5214e2c164930d Mon Sep 17 00:00:00 2001 From: dsx137 <70027572+dsx137@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:36:11 +0800 Subject: [PATCH] feat: afterMsgSaveDB webhook (#3788) * feat: afterMsgSaveDB webhook fix(config): add default webhook event filters fix(core): handle group mentions and webhook enqueue errors Use group IDs when filtering group-message attention callbacks, and report failures when asynchronous webhook tasks cannot be queued or posted. feat(webhook): add standalone mock webhook server CLI test(webhook): add mock server and test suite for all callback commands feat(msgtransfer): trigger afterMsgSaveDB webhook when message saved to DB feat(config): add afterMsgSaveDB webhook configuration feat(callbackstruct): add CallbackAfterMsgSaveDB command and request structs * refactor(core): simplify asynchronous webhook enqueueing --- .gitignore | 2 + config/webhooks.yml | 6 + internal/msgtransfer/callback.go | 73 ++++++ internal/msgtransfer/callback_test.go | 102 +++++++++ internal/msgtransfer/init.go | 2 +- .../online_msg_to_mongo_handler.go | 10 +- internal/rpc/msg/callback.go | 10 +- internal/rpc/msg/filter.go | 67 ------ pkg/callbackstruct/constant.go | 1 + pkg/callbackstruct/message.go | 10 + pkg/common/config/config.go | 1 + pkg/common/config/load_config_test.go | 3 +- pkg/common/webhook/content_type.go | 62 ++++++ pkg/common/webhook/content_type_test.go | 171 ++++++++++++++ pkg/common/webhook/http_client.go | 3 +- pkg/common/webhook/http_client_test.go | 45 ++++ test/webhook/callbacks_test.go | 209 ++++++++++++++++++ test/webhook/cmd/main.go | 103 +++++++++ test/webhook/default_handlers.go | 177 +++++++++++++++ test/webhook/server.go | 184 +++++++++++++++ test/webhook/server_test.go | 122 ++++++++++ 21 files changed, 1287 insertions(+), 76 deletions(-) create mode 100644 internal/msgtransfer/callback.go create mode 100644 internal/msgtransfer/callback_test.go delete mode 100644 internal/rpc/msg/filter.go create mode 100644 pkg/common/webhook/content_type.go create mode 100644 pkg/common/webhook/content_type_test.go create mode 100644 test/webhook/callbacks_test.go create mode 100644 test/webhook/cmd/main.go create mode 100644 test/webhook/default_handlers.go create mode 100644 test/webhook/server.go create mode 100644 test/webhook/server_test.go diff --git a/.gitignore b/.gitignore index cc9d0b04e..56b4823d5 100644 --- a/.gitignore +++ b/.gitignore @@ -388,3 +388,5 @@ Sessionx.vim # End of https://www.toptal.com/developers/gitignore/api/go,git,vim,tags,test,emacs,backup,jetbrains .idea dist/ + +.omo \ No newline at end of file diff --git a/config/webhooks.yml b/config/webhooks.yml index 854d2dc2c..4851f381c 100644 --- a/config/webhooks.yml +++ b/config/webhooks.yml @@ -49,6 +49,12 @@ afterSendGroupMsg: # See beforeSendSingleMsg comment. allowedTypes: [] deniedTypes: [] +afterMsgSaveDB: + enable: false + timeout: 5 + attentionIds: [] + allowedTypes: [] + deniedTypes: [] afterUserOnline: enable: false timeout: 5 diff --git a/internal/msgtransfer/callback.go b/internal/msgtransfer/callback.go new file mode 100644 index 000000000..95cfe9a40 --- /dev/null +++ b/internal/msgtransfer/callback.go @@ -0,0 +1,73 @@ +package msgtransfer + +import ( + "context" + + "google.golang.org/protobuf/proto" + + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" + "github.com/openimsdk/protocol/constant" + "github.com/openimsdk/protocol/sdkws" + "github.com/openimsdk/tools/mcontext" + + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" +) + +func toCommonCallback(ctx context.Context, msg *sdkws.MsgData, command string) cbapi.CommonCallbackReq { + return cbapi.CommonCallbackReq{ + SendID: msg.SendID, + ServerMsgID: msg.ServerMsgID, + CallbackCommand: command, + ClientMsgID: msg.ClientMsgID, + OperationID: mcontext.GetOperationID(ctx), + SenderPlatformID: msg.SenderPlatformID, + SenderNickname: msg.SenderNickname, + SessionType: msg.SessionType, + MsgFrom: msg.MsgFrom, + ContentType: msg.ContentType, + Status: msg.Status, + SendTime: msg.SendTime, + CreateTime: msg.CreateTime, + AtUserIDList: msg.AtUserIDList, + SenderFaceURL: msg.SenderFaceURL, + Content: GetContent(msg), + Seq: uint32(msg.Seq), + Ex: msg.Ex, + } +} + +func GetContent(msg *sdkws.MsgData) string { + if msg.ContentType >= constant.NotificationBegin && msg.ContentType <= constant.NotificationEnd { + var tips sdkws.TipsComm + _ = proto.Unmarshal(msg.Content, &tips) + content := tips.JsonDetail + return content + } else { + return string(msg.Content) + } +} + +func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterMsgSaveDB(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) { + target := msg.RecvID + if msg.SessionType == constant.ReadGroupChatType { + target = msg.GroupID + } + if !webhook.FilterAfterMsg(msg, after, target) { + return + } + + cbReq := &cbapi.CallbackAfterMsgSaveDBReq{ + CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterMsgSaveDBCommand), + } + + switch msg.SessionType { + case constant.SingleChatType, constant.NotificationChatType: + cbReq.RecvID = msg.RecvID + case constant.ReadGroupChatType: + cbReq.GroupID = msg.GroupID + default: + } + + mc.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterMsgSaveDBResp{}, after) +} diff --git a/internal/msgtransfer/callback_test.go b/internal/msgtransfer/callback_test.go new file mode 100644 index 000000000..f15c9853c --- /dev/null +++ b/internal/msgtransfer/callback_test.go @@ -0,0 +1,102 @@ +package msgtransfer + +import ( + "context" + "testing" + + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" + "github.com/openimsdk/protocol/constant" + "github.com/openimsdk/protocol/sdkws" + "github.com/stretchr/testify/assert" + "google.golang.org/protobuf/proto" + + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" +) + +func Test_FilterAfterMsg_when_configuredForTypingAndNotification(t *testing.T) { + allowed := &config.AfterConfig{AllowedTypes: []string{"0-2147483647"}} + denied := &config.AfterConfig{ + AllowedTypes: []string{"0-2147483647"}, + DeniedTypes: []string{"0-2147483647"}, + } + typingMsg := &sdkws.MsgData{RecvID: "recipient", ContentType: constant.Typing} + notificationMsg := &sdkws.MsgData{RecvID: "recipient", ContentType: constant.GroupCreatedNotification} + + assert.True(t, webhook.FilterAfterMsg(typingMsg, allowed, "recipient")) + assert.True(t, webhook.FilterAfterMsg(notificationMsg, allowed, "recipient")) + assert.False(t, webhook.FilterAfterMsg(typingMsg, denied, "recipient")) + assert.False(t, webhook.FilterAfterMsg(notificationMsg, denied, "recipient")) +} + +func Test_FilterAfterMsg_when_customAllowedAndDeniedIntervals(t *testing.T) { + // Given: after-config with allowed and denied intervals + afterConf := &config.AfterConfig{ + AllowedTypes: []string{"101-105", "201"}, + DeniedTypes: []string{"103"}, + } + + // When & Then: test matching allowed + assert.True(t, webhook.FilterAfterMsg(&sdkws.MsgData{RecvID: "recipient", ContentType: 101}, afterConf, "recipient")) + assert.True(t, webhook.FilterAfterMsg(&sdkws.MsgData{RecvID: "recipient", ContentType: 201}, afterConf, "recipient")) + + // When & Then: test matching denied (103 is in 101-105, but also in denied 103) + assert.False(t, webhook.FilterAfterMsg(&sdkws.MsgData{RecvID: "recipient", ContentType: 103}, afterConf, "recipient")) + + // When & Then: test not in allowed (300) + assert.False(t, webhook.FilterAfterMsg(&sdkws.MsgData{RecvID: "recipient", ContentType: 300}, afterConf, "recipient")) +} + +func Test_FilterAfterMsg_when_attentionIDsConfigured(t *testing.T) { + groupMsg := &sdkws.MsgData{ + SendID: "sender", + RecvID: "recipient", + GroupID: "group", + SessionType: constant.ReadGroupChatType, + ContentType: constant.Picture, + } + + // When: attention matches the sender + assert.True(t, webhook.FilterAfterMsg(groupMsg, &config.AfterConfig{AttentionIds: []string{"sender"}}, "group")) + + // When: attention matches the group target + assert.True(t, webhook.FilterAfterMsg(groupMsg, &config.AfterConfig{AttentionIds: []string{"group"}}, "group")) + + // Then: attention does not match the recipient for a group message + assert.False(t, webhook.FilterAfterMsg(groupMsg, &config.AfterConfig{AttentionIds: []string{"recipient"}}, "group")) +} + +func Test_ToCommonCallback_and_GetContent(t *testing.T) { + // Given: a text message and a tips message + ctx := context.Background() + textMsg := &sdkws.MsgData{ + SendID: "sender_1", + RecvID: "recv_1", + ServerMsgID: "server_msg_1", + ClientMsgID: "client_msg_1", + ContentType: constant.Picture, + Content: []byte("https://example.com/pic.jpg"), + } + + tips := &sdkws.TipsComm{JsonDetail: `{"detail":"group created"}`} + tipsBytes, err := proto.Marshal(tips) + assert.NoError(t, err) + notiMsg := &sdkws.MsgData{ + SendID: "sender_1", + GroupID: "group_1", + ServerMsgID: "server_msg_2", + ClientMsgID: "client_msg_2", + ContentType: constant.GroupCreatedNotification, + Content: tipsBytes, + } + + // When: converting to common callback + textCb := toCommonCallback(ctx, textMsg, cbapi.CallbackAfterMsgSaveDBCommand) + notiContent := GetContent(notiMsg) + + // Then: contents and fields are correctly mapped + assert.Equal(t, "sender_1", textCb.SendID) + assert.Equal(t, "https://example.com/pic.jpg", textCb.Content) + assert.Equal(t, cbapi.CallbackAfterMsgSaveDBCommand, textCb.CallbackCommand) + assert.Equal(t, `{"detail":"group created"}`, notiContent) +} diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go index be858ace1..821bd012c 100644 --- a/internal/msgtransfer/init.go +++ b/internal/msgtransfer/init.go @@ -112,7 +112,7 @@ func Start(ctx context.Context, index int, config *Config) error { if err != nil { return err } - historyMongoCH, err := NewOnlineHistoryMongoConsumerHandler(&config.KafkaConfig, msgTransferDatabase) + historyMongoCH, err := NewOnlineHistoryMongoConsumerHandler(&config.KafkaConfig, msgTransferDatabase, config) if err != nil { return err } diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index 01872562e..cdf2f381f 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -22,6 +22,7 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" + "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" pbmsg "github.com/openimsdk/protocol/msg" "github.com/openimsdk/tools/log" "google.golang.org/protobuf/proto" @@ -30,9 +31,11 @@ import ( type OnlineHistoryMongoConsumerHandler struct { historyConsumerGroup *kafka.MConsumerGroup msgTransferDatabase controller.MsgTransferDatabase + config *Config + webhookClient *webhook.Client } -func NewOnlineHistoryMongoConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase) (*OnlineHistoryMongoConsumerHandler, error) { +func NewOnlineHistoryMongoConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase, config *Config) (*OnlineHistoryMongoConsumerHandler, error) { historyConsumerGroup, err := kafka.NewMConsumerGroup(kafkaConf.Build(), kafkaConf.ToMongoGroupID, []string{kafkaConf.ToMongoTopic}, true) if err != nil { return nil, err @@ -41,6 +44,8 @@ func NewOnlineHistoryMongoConsumerHandler(kafkaConf *config.Kafka, database cont mc := &OnlineHistoryMongoConsumerHandler{ historyConsumerGroup: historyConsumerGroup, msgTransferDatabase: database, + config: config, + webhookClient: webhook.NewWebhookClient(config.WebhooksConfig.URL), } return mc, nil } @@ -72,6 +77,9 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont prommetrics.MsgInsertMongoFailedCounter.Inc() } else { prommetrics.MsgInsertMongoSuccessCounter.Inc() + for _, msgData := range msgFromMQ.MsgData { + mc.webhookAfterMsgSaveDB(ctx, &mc.config.WebhooksConfig.AfterMsgSaveDB, msgData) + } } //var seqs []int64 //for _, msg := range msgFromMQ.MsgData { diff --git a/internal/rpc/msg/callback.go b/internal/rpc/msg/callback.go index c66dd6ca9..993de6dab 100644 --- a/internal/rpc/msg/callback.go +++ b/internal/rpc/msg/callback.go @@ -67,7 +67,7 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf if msg.MsgData.ContentType == constant.Typing { return nil } - if !filterBeforeMsg(msg, before) { + if !webhook.FilterBeforeMsg(msg.MsgData, before) { return nil } cbReq := &cbapi.CallbackBeforeSendSingleMsgReq{ @@ -87,7 +87,7 @@ func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config if msg.MsgData.ContentType == constant.Typing { return } - if !filterAfterMsg(msg, after) { + if !webhook.FilterAfterMsg(msg.MsgData, after, msg.MsgData.RecvID) { return } cbReq := &cbapi.CallbackAfterSendSingleMsgReq{ @@ -99,7 +99,7 @@ func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error { return webhook.WithCondition(ctx, before, func(ctx context.Context) error { - if !filterBeforeMsg(msg, before) { + if !webhook.FilterBeforeMsg(msg.MsgData, before) { return nil } if msg.MsgData.ContentType == constant.Typing { @@ -121,7 +121,7 @@ func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config. if msg.MsgData.ContentType == constant.Typing { return } - if !filterAfterMsg(msg, after) { + if !webhook.FilterAfterMsg(msg.MsgData, after, msg.MsgData.RecvID) { return } cbReq := &cbapi.CallbackAfterSendGroupMsgReq{ @@ -136,7 +136,7 @@ func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.B if msg.MsgData.ContentType != constant.Text { return nil } - if !filterBeforeMsg(msg, before) { + if !webhook.FilterBeforeMsg(msg.MsgData, before) { return nil } cbReq := &cbapi.CallbackMsgModifyCommandReq{ diff --git a/internal/rpc/msg/filter.go b/internal/rpc/msg/filter.go deleted file mode 100644 index ed1a488f1..000000000 --- a/internal/rpc/msg/filter.go +++ /dev/null @@ -1,67 +0,0 @@ -package msg - -import ( - "github.com/openimsdk/open-im-server/v3/pkg/common/config" - pbchat "github.com/openimsdk/protocol/msg" - "github.com/openimsdk/tools/utils/datautil" - "strconv" - "strings" -) - -const ( - separator = "-" -) - -func filterAfterMsg(msg *pbchat.SendMsgReq, after *config.AfterConfig) bool { - return filterMsg(msg, after.AttentionIds, after.AllowedTypes, after.DeniedTypes) -} - -func filterBeforeMsg(msg *pbchat.SendMsgReq, before *config.BeforeConfig) bool { - return filterMsg(msg, nil, before.AllowedTypes, before.DeniedTypes) -} - -func filterMsg(msg *pbchat.SendMsgReq, attentionIds, allowedTypes, deniedTypes []string) bool { - // According to the attentionIds configuration, only some users are sent - if len(attentionIds) != 0 && !datautil.Contains([]string{msg.MsgData.SendID, msg.MsgData.RecvID}, attentionIds...) { - return false - } - if len(allowedTypes) != 0 && !isInInterval(msg.MsgData.ContentType, allowedTypes) { - return false - } - if len(deniedTypes) != 0 && isInInterval(msg.MsgData.ContentType, deniedTypes) { - return false - } - return true -} - -func isInInterval(contentType int32, interval []string) bool { - for _, v := range interval { - if strings.Contains(v, separator) { - // is interval - bounds := strings.Split(v, separator) - if len(bounds) != 2 { - continue - } - bottom, err := strconv.Atoi(bounds[0]) - if err != nil { - continue - } - top, err := strconv.Atoi(bounds[1]) - if err != nil { - continue - } - if datautil.BetweenEq(int(contentType), bottom, top) { - return true - } - } else { - iv, err := strconv.Atoi(v) - if err != nil { - continue - } - if int(contentType) == iv { - return true - } - } - } - return false -} diff --git a/pkg/callbackstruct/constant.go b/pkg/callbackstruct/constant.go index 73f89a719..d8d2bde34 100644 --- a/pkg/callbackstruct/constant.go +++ b/pkg/callbackstruct/constant.go @@ -62,4 +62,5 @@ const ( CallbackBeforeMembersJoinGroupCommand = "callbackBeforeMembersJoinGroupCommand" CallbackBeforeSetGroupMemberInfoCommand = "callbackBeforeSetGroupMemberInfoCommand" CallbackAfterSetGroupMemberInfoCommand = "callbackAfterSetGroupMemberInfoCommand" + CallbackAfterMsgSaveDBCommand = "callbackAfterMsgSaveDBCommand" ) diff --git a/pkg/callbackstruct/message.go b/pkg/callbackstruct/message.go index 902fa6110..ef8a8bb28 100644 --- a/pkg/callbackstruct/message.go +++ b/pkg/callbackstruct/message.go @@ -103,3 +103,13 @@ type CallbackSingleMsgReadReq struct { type CallbackSingleMsgReadResp struct { CommonCallbackResp } + +type CallbackAfterMsgSaveDBReq struct { + CommonCallbackReq + RecvID string `json:"recvID"` + GroupID string `json:"groupID"` +} + +type CallbackAfterMsgSaveDBResp struct { + CommonCallbackResp +} diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index 7e54bf4d4..f7922bb51 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -433,6 +433,7 @@ type Webhooks struct { BeforeSendGroupMsg BeforeConfig `mapstructure:"beforeSendGroupMsg"` BeforeMsgModify BeforeConfig `mapstructure:"beforeMsgModify"` AfterSendGroupMsg AfterConfig `mapstructure:"afterSendGroupMsg"` + AfterMsgSaveDB AfterConfig `mapstructure:"afterMsgSaveDB"` AfterUserOnline AfterConfig `mapstructure:"afterUserOnline"` AfterUserOffline AfterConfig `mapstructure:"afterUserOffline"` AfterUserKickOff AfterConfig `mapstructure:"afterUserKickOff"` diff --git a/pkg/common/config/load_config_test.go b/pkg/common/config/load_config_test.go index fab85c3a6..4123c5162 100644 --- a/pkg/common/config/load_config_test.go +++ b/pkg/common/config/load_config_test.go @@ -28,7 +28,8 @@ func TestLoadWebhooksConfig(t *testing.T) { err := LoadConfig("../../../config/webhooks.yml", "IMENV_WEBHOOKS", &webhooks) assert.Nil(t, err) assert.Equal(t, 5, webhooks.BeforeAddBlack.Timeout) - + assert.Equal(t, 5, webhooks.AfterMsgSaveDB.Timeout) + assert.Equal(t, false, webhooks.AfterMsgSaveDB.Enable) } func TestLoadDiscoveryKubernetesConfig(t *testing.T) { diff --git a/pkg/common/webhook/content_type.go b/pkg/common/webhook/content_type.go new file mode 100644 index 000000000..da893e993 --- /dev/null +++ b/pkg/common/webhook/content_type.go @@ -0,0 +1,62 @@ +package webhook + +import ( + "strconv" + "strings" + + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/protocol/sdkws" + "github.com/openimsdk/tools/utils/datautil" +) + +func IsContentTypeInIntervals(contentType int32, intervals []string) bool { + for _, interval := range intervals { + if strings.Contains(interval, "-") { + bounds := strings.Split(interval, "-") + if len(bounds) != 2 { + continue + } + bottom, err := strconv.Atoi(bounds[0]) + if err != nil { + continue + } + top, err := strconv.Atoi(bounds[1]) + if err != nil { + continue + } + if datautil.BetweenEq(int(contentType), bottom, top) { + return true + } + } else { + contentTypeValue, err := strconv.Atoi(interval) + if err != nil { + continue + } + if int(contentType) == contentTypeValue { + return true + } + } + } + return false +} + +func FilterBeforeMsg(msg *sdkws.MsgData, before *config.BeforeConfig) bool { + return filterContentType(msg.ContentType, before.AllowedTypes, before.DeniedTypes) +} + +func FilterAfterMsg(msg *sdkws.MsgData, after *config.AfterConfig, attentionTargetID string) bool { + if len(after.AttentionIds) != 0 && !datautil.Contains([]string{msg.SendID, attentionTargetID}, after.AttentionIds...) { + return false + } + return filterContentType(msg.ContentType, after.AllowedTypes, after.DeniedTypes) +} + +func filterContentType(contentType int32, allowedTypes, deniedTypes []string) bool { + if len(allowedTypes) != 0 && !IsContentTypeInIntervals(contentType, allowedTypes) { + return false + } + if len(deniedTypes) != 0 && IsContentTypeInIntervals(contentType, deniedTypes) { + return false + } + return true +} diff --git a/pkg/common/webhook/content_type_test.go b/pkg/common/webhook/content_type_test.go new file mode 100644 index 000000000..93715c09b --- /dev/null +++ b/pkg/common/webhook/content_type_test.go @@ -0,0 +1,171 @@ +package webhook + +import ( + "testing" + + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/protocol/constant" + "github.com/openimsdk/protocol/sdkws" +) + +func TestFilterAfterMsg(t *testing.T) { + tests := []struct { + name string + msg *sdkws.MsgData + after *config.AfterConfig + attentionTarget string + want bool + }{ + { + name: "empty filters allow message", + msg: &sdkws.MsgData{SendID: "sender", ContentType: 1}, + after: &config.AfterConfig{}, + want: true, + }, + { + name: "attention matches sender", + msg: &sdkws.MsgData{SendID: "sender", ContentType: 1}, + after: &config.AfterConfig{AttentionIds: []string{"sender"}}, + attentionTarget: "other", + want: true, + }, + { + name: "attention matches explicit target", + msg: &sdkws.MsgData{SendID: "sender", ContentType: 1}, + after: &config.AfterConfig{AttentionIds: []string{"target"}}, + attentionTarget: "target", + want: true, + }, + { + name: "attention rejects unrelated ids", + msg: &sdkws.MsgData{SendID: "sender", ContentType: 1}, + after: &config.AfterConfig{AttentionIds: []string{"unrelated"}}, + attentionTarget: "target", + want: false, + }, + { + name: "allowed type hit", + msg: &sdkws.MsgData{ContentType: 7}, + after: &config.AfterConfig{AllowedTypes: []string{"1-7"}}, + want: true, + }, + { + name: "allowed type miss", + msg: &sdkws.MsgData{ContentType: 8}, + after: &config.AfterConfig{AllowedTypes: []string{"1-7"}}, + want: false, + }, + { + name: "denied type hit", + msg: &sdkws.MsgData{ContentType: 8}, + after: &config.AfterConfig{DeniedTypes: []string{"8"}}, + want: false, + }, + { + name: "denied wins over allowed", + msg: &sdkws.MsgData{ContentType: 8}, + after: &config.AfterConfig{ + AllowedTypes: []string{"1-10"}, + DeniedTypes: []string{"8"}, + }, + want: false, + }, + { + name: "malformed allowed interval rejects", + msg: &sdkws.MsgData{ContentType: 8}, + after: &config.AfterConfig{AllowedTypes: []string{"bad-interval"}}, + want: false, + }, + { + name: "malformed denied interval does not reject", + msg: &sdkws.MsgData{ContentType: 8}, + after: &config.AfterConfig{DeniedTypes: []string{"bad-interval"}}, + want: true, + }, + { + name: "typing is not hardcoded denied", + msg: &sdkws.MsgData{ContentType: constant.Typing}, + after: &config.AfterConfig{AllowedTypes: []string{"113"}}, + want: true, + }, + { + name: "notification is not hardcoded denied", + msg: &sdkws.MsgData{ContentType: constant.GroupCreatedNotification}, + after: &config.AfterConfig{ + AllowedTypes: []string{"1501"}, + }, + want: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := FilterAfterMsg(test.msg, test.after, test.attentionTarget); got != test.want { + t.Fatalf("FilterAfterMsg() = %v, want %v", got, test.want) + } + }) + } +} + +func TestFilterBeforeMsg(t *testing.T) { + tests := []struct { + name string + msg *sdkws.MsgData + before *config.BeforeConfig + want bool + }{ + { + name: "empty filters allow message", + msg: &sdkws.MsgData{ContentType: 1}, + before: &config.BeforeConfig{}, + want: true, + }, + { + name: "allowed type hit", + msg: &sdkws.MsgData{ContentType: 7}, + before: &config.BeforeConfig{AllowedTypes: []string{"1-7"}}, + want: true, + }, + { + name: "allowed type miss", + msg: &sdkws.MsgData{ContentType: 8}, + before: &config.BeforeConfig{AllowedTypes: []string{"1-7"}}, + want: false, + }, + { + name: "denied type hit", + msg: &sdkws.MsgData{ContentType: 8}, + before: &config.BeforeConfig{DeniedTypes: []string{"8"}}, + want: false, + }, + { + name: "denied wins over allowed", + msg: &sdkws.MsgData{ContentType: 8}, + before: &config.BeforeConfig{ + AllowedTypes: []string{"1-10"}, + DeniedTypes: []string{"8"}, + }, + want: false, + }, + { + name: "malformed allowed interval rejects", + msg: &sdkws.MsgData{ContentType: 8}, + before: &config.BeforeConfig{AllowedTypes: []string{"bad-interval"}}, + want: false, + }, + { + name: "malformed denied interval does not reject", + msg: &sdkws.MsgData{ContentType: 8}, + before: &config.BeforeConfig{DeniedTypes: []string{"bad-interval"}}, + want: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := FilterBeforeMsg(test.msg, test.before); got != test.want { + t.Fatalf("FilterBeforeMsg() = %v, want %v", got, test.want) + } + }) + } +} diff --git a/pkg/common/webhook/http_client.go b/pkg/common/webhook/http_client.go index e46f08806..3051c41c8 100644 --- a/pkg/common/webhook/http_client.go +++ b/pkg/common/webhook/http_client.go @@ -17,6 +17,8 @@ package webhook import ( "context" "encoding/json" + "net/http" + "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" @@ -25,7 +27,6 @@ import ( "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mq/memamq" "github.com/openimsdk/tools/utils/httputil" - "net/http" ) type Client struct { diff --git a/pkg/common/webhook/http_client_test.go b/pkg/common/webhook/http_client_test.go index 3c3aeb809..06ed37745 100644 --- a/pkg/common/webhook/http_client_test.go +++ b/pkg/common/webhook/http_client_test.go @@ -13,3 +13,48 @@ // limitations under the License. package webhook + +import ( + "context" + "testing" + "time" + + "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/tools/mq/memamq" +) + +func TestAsyncPost_whenQueueFull_returnsWithoutWaiting(t *testing.T) { + queue := memamq.NewMemoryQueue(1, 1) + taskStarted := make(chan struct{}) + release := make(chan struct{}) + t.Cleanup(func() { + close(release) + queue.Stop() + }) + + if err := queue.Push(func() { + close(taskStarted) + <-release + }); err != nil { + t.Fatal(err) + } + <-taskStarted + if err := queue.Push(func() {}); err != nil { + t.Fatal(err) + } + + client := NewWebhookClient("http://example.com", queue) + started := time.Now() + client.AsyncPost( + context.Background(), + "command", + &callbackstruct.CommonCallbackReq{CallbackCommand: "command"}, + &callbackstruct.CommonCallbackResp{}, + &config.AfterConfig{Enable: true}, + ) + + if elapsed := time.Since(started); elapsed >= time.Second { + t.Fatalf("AsyncPost waited %s for queue capacity", elapsed) + } +} diff --git a/test/webhook/callbacks_test.go b/test/webhook/callbacks_test.go new file mode 100644 index 000000000..37193c0f1 --- /dev/null +++ b/test/webhook/callbacks_test.go @@ -0,0 +1,209 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package webhook + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" + sdkws "github.com/openimsdk/protocol/sdkws" + "github.com/openimsdk/protocol/wrapperspb" + "github.com/stretchr/testify/require" +) + +func runWebhookTest(t *testing.T, srv *Server, command string, reqObj any) *httptest.ResponseRecorder { + t.Helper() + body, err := json.Marshal(reqObj) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, "/callbackExample/"+command, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + srv.ServeHTTP(w, req) + require.Equal(t, http.StatusOK, w.Code) + return w +} + +func TestAllWebhookCallbacks_UserAndStatus(t *testing.T) { + t.Parallel() + srv := NewServer() + + t.Run("UserRegister", func(t *testing.T) { + w := runWebhookTest(t, srv, cbapi.CallbackBeforeUserRegisterCommand, &cbapi.CallbackBeforeUserRegisterReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackBeforeUserRegisterCommand), + Users: []*sdkws.UserInfo{{UserID: "u1", Nickname: "User1"}}, + }) + var resp cbapi.CallbackBeforeUserRegisterResp + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + require.Len(t, resp.Users, 1) + require.Equal(t, "User1", resp.Users[0].Nickname) + + runWebhookTest(t, srv, cbapi.CallbackAfterUserRegisterCommand, &cbapi.CallbackAfterUserRegisterReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackAfterUserRegisterCommand), + Users: []*sdkws.UserInfo{{UserID: "u1"}}, + }) + }) + + t.Run("UpdateUserInfo", func(t *testing.T) { + newNick := "NewNick" + w := runWebhookTest(t, srv, cbapi.CallbackBeforeUpdateUserInfoCommand, &cbapi.CallbackBeforeUpdateUserInfoReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackBeforeUpdateUserInfoCommand), + UserID: "u1", + Nickname: &newNick, + }) + var resp cbapi.CallbackBeforeUpdateUserInfoResp + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + require.NotNil(t, resp.Nickname) + require.Equal(t, "NewNick", *resp.Nickname) + + runWebhookTest(t, srv, cbapi.CallbackAfterUpdateUserInfoCommand, &cbapi.CallbackAfterUpdateUserInfoReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackAfterUpdateUserInfoCommand), + UserID: "u1", + Nickname: "NewNick", + }) + }) + + t.Run("UpdateUserInfoEx", func(t *testing.T) { + w := runWebhookTest(t, srv, cbapi.CallbackBeforeUpdateUserInfoExCommand, &cbapi.CallbackBeforeUpdateUserInfoExReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackBeforeUpdateUserInfoExCommand), + UserID: "u1", + Ex: wrapperspb.String("new-ex"), + }) + var resp cbapi.CallbackBeforeUpdateUserInfoExResp + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + require.NotNil(t, resp.Ex) + require.Equal(t, "new-ex", resp.Ex.Value) + + runWebhookTest(t, srv, cbapi.CallbackAfterUpdateUserInfoExCommand, &cbapi.CallbackAfterUpdateUserInfoExReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackAfterUpdateUserInfoExCommand), + UserID: "u1", + Ex: wrapperspb.String("new-ex"), + }) + }) + + t.Run("UserStatusGateway", func(t *testing.T) { + for _, cmd := range []string{ + cbapi.CallbackAfterUserOnlineCommand, + cbapi.CallbackAfterUserOfflineCommand, + cbapi.CallbackAfterUserKickOffCommand, + } { + runWebhookTest(t, srv, cmd, &cbapi.CallbackUserOnlineReq{ + UserStatusCallbackReq: cbapi.UserStatusCallbackReq{ + UserStatusBaseCallback: cbapi.UserStatusBaseCallback{ + CallbackCommand: cmd, + Platform: "iOS", + }, + UserID: "u1", + }, + }) + } + }) +} + +func TestAllWebhookCallbacks_FriendAndGroup(t *testing.T) { + t.Parallel() + srv := NewServer() + + t.Run("FriendCallbacks", func(t *testing.T) { + friendCmds := []string{ + cbapi.CallbackBeforeAddFriendCommand, + cbapi.CallbackAfterAddFriendCommand, + cbapi.CallbackBeforeAddFriendAgreeCommand, + cbapi.CallbackAfterAddFriendAgreeCommand, + cbapi.CallbackAfterDeleteFriendCommand, + cbapi.CallbackBeforeImportFriendsCommand, + cbapi.CallbackAfterImportFriendsCommand, + cbapi.CallbackBeforeSetFriendRemarkCommand, + cbapi.CallbackAfterSetFriendRemarkCommand, + cbapi.CallbackBeforeAddBlackCommand, + cbapi.CallbackAfterRemoveBlackCommand, + } + for _, cmd := range friendCmds { + runWebhookTest(t, srv, cmd, &cbapi.CallbackBeforeAddFriendReq{ + CallbackCommand: cbapi.CallbackCommand(cmd), + FromUserID: "u1", + ToUserID: "u2", + }) + } + }) + + t.Run("GroupCallbacks", func(t *testing.T) { + groupCmds := []string{ + cbapi.CallbackBeforeCreateGroupCommand, + cbapi.CallbackAfterCreateGroupCommand, + cbapi.CallbackBeforeMembersJoinGroupCommand, + cbapi.CallbackBeforeJoinGroupCommand, + cbapi.CallbackAfterJoinGroupCommand, + cbapi.CallbackAfterQuitGroupCommand, + cbapi.CallbackAfterKickGroupCommand, + cbapi.CallbackAfterDisMissGroupCommand, + cbapi.CallbackAfterTransferGroupOwnerCommand, + cbapi.CallbackBeforeInviteJoinGroupCommand, + cbapi.CallbackBeforeSetGroupInfoCommand, + cbapi.CallbackAfterSetGroupInfoCommand, + cbapi.CallbackBeforeSetGroupInfoExCommand, + cbapi.CallbackAfterSetGroupInfoExCommand, + cbapi.CallbackBeforeSetGroupMemberInfoCommand, + cbapi.CallbackAfterSetGroupMemberInfoCommand, + } + for _, cmd := range groupCmds { + runWebhookTest(t, srv, cmd, &cbapi.CallbackQuitGroupReq{ + CallbackCommand: cbapi.CallbackCommand(cmd), + GroupID: "g1", + UserID: "u1", + }) + } + }) +} + +func TestAllWebhookCallbacks_MsgAndPush(t *testing.T) { + t.Parallel() + srv := NewServer() + + msgCmds := []string{ + cbapi.CallbackBeforeSendSingleMsgCommand, + cbapi.CallbackAfterSendSingleMsgCommand, + cbapi.CallbackBeforeSendGroupMsgCommand, + cbapi.CallbackAfterSendGroupMsgCommand, + cbapi.CallbackBeforeMsgModifyCommand, + cbapi.CallbackAfterSingleMsgReadCommand, + cbapi.CallbackAfterGroupMsgReadCommand, + cbapi.CallbackAfterMsgSaveDBCommand, + cbapi.CallbackAfterRevokeMsgCommand, + cbapi.CallbackBeforeOfflinePushCommand, + cbapi.CallbackBeforeOnlinePushCommand, + cbapi.CallbackBeforeGroupOnlinePushCommand, + } + + for _, cmd := range msgCmds { + cmd := cmd + t.Run(cmd, func(t *testing.T) { + t.Parallel() + runWebhookTest(t, srv, cmd, &cbapi.CallbackAfterMsgSaveDBReq{ + CommonCallbackReq: cbapi.CommonCallbackReq{ + SendID: "u1", + CallbackCommand: cmd, + }, + RecvID: "u2", + GroupID: "g1", + }) + }) + } +} diff --git a/test/webhook/cmd/main.go b/test/webhook/cmd/main.go new file mode 100644 index 000000000..d9ab61666 --- /dev/null +++ b/test/webhook/cmd/main.go @@ -0,0 +1,103 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/gin-gonic/gin" + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" + "github.com/openimsdk/open-im-server/v3/test/webhook" + "github.com/openimsdk/protocol/constant" +) + +func main() { + port := flag.Int("port", 10006, "Webhook server listening port") + sensitiveWord := flag.String("mask-word", "xxx", "Keyword to mask in callbackBeforeMsgModifyCommand") + flag.Parse() + + gin.SetMode(gin.ReleaseMode) + srv := webhook.NewServer() + + // 1. Log every incoming webhook request in real-time + srv.OnRequest(func(command string, reqBody []byte, c *gin.Context) { + log.Printf("[Webhook] %s %s | Command: %s | Body: %s", + c.Request.Method, c.Request.URL.Path, command, string(reqBody)) + }) + + // 2. Example: Custom hook for message content modification (sensitive word filtering) + if *sensitiveWord != "" { + webhook.RegisterCallback(srv, cbapi.CallbackBeforeMsgModifyCommand, func(c *gin.Context, req *cbapi.CallbackMsgModifyCommandReq) (*cbapi.CallbackMsgModifyCommandResp, error) { + var resp cbapi.CallbackMsgModifyCommandResp + if req.ContentType != constant.Text { + return &resp, nil + } + var textElem struct { + Content string `json:"content"` + } + if err := json.Unmarshal([]byte(req.Content), &textElem); err != nil { + return nil, err + } + if strings.Contains(textElem.Content, *sensitiveWord) { + textElem.Content = strings.ReplaceAll(textElem.Content, *sensitiveWord, strings.Repeat("*", len(*sensitiveWord))) + masked, err := json.Marshal(&textElem) + if err != nil { + return nil, err + } + s := string(masked) + resp.Content = &s + log.Printf("[Webhook Mask] Masked content: %s -> %s", req.Content, s) + } + return &resp, nil + }) + } + + addr := fmt.Sprintf(":%d", *port) + log.Printf("=================================================================") + log.Printf(" OpenIM Mock Webhook Server is listening on %s", addr) + log.Printf(" Callback URL in config/webhooks.yml:") + log.Printf(" url: 'http://127.0.0.1:%d/callbackExample'", *port) + log.Printf(" Supported callbacks: 48 registered OpenIM callback commands") + log.Printf(" Press Ctrl+C to stop") + log.Printf("=================================================================") + + go func() { + if err := srv.Start(addr); err != nil && err != http.ErrServerClosed { + log.Fatalf("Server listen error: %v", err) + } + }() + + quit := make(chan os.Signal, 1) + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + <-quit + log.Println("Shutting down Webhook Server...") + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := srv.Shutdown(ctx); err != nil { + log.Fatalf("Server forced to shutdown: %v", err) + } + log.Println("Webhook Server exited cleanly.") +} diff --git a/test/webhook/default_handlers.go b/test/webhook/default_handlers.go new file mode 100644 index 000000000..d7624ca0b --- /dev/null +++ b/test/webhook/default_handlers.go @@ -0,0 +1,177 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package webhook + +import ( + "github.com/gin-gonic/gin" + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" +) + +// RegisterDefaultHandlers registers default success handlers for all supported OpenIM webhook commands. +func (s *Server) RegisterDefaultHandlers() { + // User + RegisterCallback[cbapi.CallbackBeforeUserRegisterReq, cbapi.CallbackBeforeUserRegisterResp](s, cbapi.CallbackBeforeUserRegisterCommand, func(c *gin.Context, req *cbapi.CallbackBeforeUserRegisterReq) (*cbapi.CallbackBeforeUserRegisterResp, error) { + return &cbapi.CallbackBeforeUserRegisterResp{Users: req.Users}, nil + }) + RegisterCallback[cbapi.CallbackAfterUserRegisterReq, cbapi.CallbackAfterUserRegisterResp](s, cbapi.CallbackAfterUserRegisterCommand, func(c *gin.Context, req *cbapi.CallbackAfterUserRegisterReq) (*cbapi.CallbackAfterUserRegisterResp, error) { + return &cbapi.CallbackAfterUserRegisterResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeUpdateUserInfoReq, cbapi.CallbackBeforeUpdateUserInfoResp](s, cbapi.CallbackBeforeUpdateUserInfoCommand, func(c *gin.Context, req *cbapi.CallbackBeforeUpdateUserInfoReq) (*cbapi.CallbackBeforeUpdateUserInfoResp, error) { + return &cbapi.CallbackBeforeUpdateUserInfoResp{Nickname: req.Nickname, FaceURL: req.FaceURL, Ex: req.Ex}, nil + }) + RegisterCallback[cbapi.CallbackAfterUpdateUserInfoReq, cbapi.CallbackAfterUpdateUserInfoResp](s, cbapi.CallbackAfterUpdateUserInfoCommand, func(c *gin.Context, req *cbapi.CallbackAfterUpdateUserInfoReq) (*cbapi.CallbackAfterUpdateUserInfoResp, error) { + return &cbapi.CallbackAfterUpdateUserInfoResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeUpdateUserInfoExReq, cbapi.CallbackBeforeUpdateUserInfoExResp](s, cbapi.CallbackBeforeUpdateUserInfoExCommand, func(c *gin.Context, req *cbapi.CallbackBeforeUpdateUserInfoExReq) (*cbapi.CallbackBeforeUpdateUserInfoExResp, error) { + return &cbapi.CallbackBeforeUpdateUserInfoExResp{Nickname: req.Nickname, FaceURL: req.FaceURL, Ex: req.Ex}, nil + }) + RegisterCallback[cbapi.CallbackAfterUpdateUserInfoExReq, cbapi.CallbackAfterUpdateUserInfoExResp](s, cbapi.CallbackAfterUpdateUserInfoExCommand, func(c *gin.Context, req *cbapi.CallbackAfterUpdateUserInfoExReq) (*cbapi.CallbackAfterUpdateUserInfoExResp, error) { + return &cbapi.CallbackAfterUpdateUserInfoExResp{}, nil + }) + + // User Status Gateway + RegisterCallback[cbapi.CallbackUserOnlineReq, cbapi.CallbackUserOnlineResp](s, cbapi.CallbackAfterUserOnlineCommand, func(c *gin.Context, req *cbapi.CallbackUserOnlineReq) (*cbapi.CallbackUserOnlineResp, error) { + return &cbapi.CallbackUserOnlineResp{}, nil + }) + RegisterCallback[cbapi.CallbackUserOfflineReq, cbapi.CallbackUserOfflineResp](s, cbapi.CallbackAfterUserOfflineCommand, func(c *gin.Context, req *cbapi.CallbackUserOfflineReq) (*cbapi.CallbackUserOfflineResp, error) { + return &cbapi.CallbackUserOfflineResp{}, nil + }) + RegisterCallback[cbapi.CallbackUserKickOffReq, cbapi.CallbackUserKickOffResp](s, cbapi.CallbackAfterUserKickOffCommand, func(c *gin.Context, req *cbapi.CallbackUserKickOffReq) (*cbapi.CallbackUserKickOffResp, error) { + return &cbapi.CallbackUserKickOffResp{}, nil + }) + + // Friend + RegisterCallback[cbapi.CallbackBeforeAddFriendReq, cbapi.CallbackBeforeAddFriendResp](s, cbapi.CallbackBeforeAddFriendCommand, func(c *gin.Context, req *cbapi.CallbackBeforeAddFriendReq) (*cbapi.CallbackBeforeAddFriendResp, error) { + return &cbapi.CallbackBeforeAddFriendResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterAddFriendReq, cbapi.CallbackAfterAddFriendResp](s, cbapi.CallbackAfterAddFriendCommand, func(c *gin.Context, req *cbapi.CallbackAfterAddFriendReq) (*cbapi.CallbackAfterAddFriendResp, error) { + return &cbapi.CallbackAfterAddFriendResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeAddFriendAgreeReq, cbapi.CallbackBeforeAddFriendAgreeResp](s, cbapi.CallbackBeforeAddFriendAgreeCommand, func(c *gin.Context, req *cbapi.CallbackBeforeAddFriendAgreeReq) (*cbapi.CallbackBeforeAddFriendAgreeResp, error) { + return &cbapi.CallbackBeforeAddFriendAgreeResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterAddFriendAgreeReq, cbapi.CallbackAfterAddFriendAgreeResp](s, cbapi.CallbackAfterAddFriendAgreeCommand, func(c *gin.Context, req *cbapi.CallbackAfterAddFriendAgreeReq) (*cbapi.CallbackAfterAddFriendAgreeResp, error) { + return &cbapi.CallbackAfterAddFriendAgreeResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterDeleteFriendReq, cbapi.CallbackAfterDeleteFriendResp](s, cbapi.CallbackAfterDeleteFriendCommand, func(c *gin.Context, req *cbapi.CallbackAfterDeleteFriendReq) (*cbapi.CallbackAfterDeleteFriendResp, error) { + return &cbapi.CallbackAfterDeleteFriendResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeImportFriendsReq, cbapi.CallbackBeforeImportFriendsResp](s, cbapi.CallbackBeforeImportFriendsCommand, func(c *gin.Context, req *cbapi.CallbackBeforeImportFriendsReq) (*cbapi.CallbackBeforeImportFriendsResp, error) { + return &cbapi.CallbackBeforeImportFriendsResp{FriendUserIDs: req.FriendUserIDs}, nil + }) + RegisterCallback[cbapi.CallbackAfterImportFriendsReq, cbapi.CallbackAfterImportFriendsResp](s, cbapi.CallbackAfterImportFriendsCommand, func(c *gin.Context, req *cbapi.CallbackAfterImportFriendsReq) (*cbapi.CallbackAfterImportFriendsResp, error) { + return &cbapi.CallbackAfterImportFriendsResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeSetFriendRemarkReq, cbapi.CallbackBeforeSetFriendRemarkResp](s, cbapi.CallbackBeforeSetFriendRemarkCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSetFriendRemarkReq) (*cbapi.CallbackBeforeSetFriendRemarkResp, error) { + return &cbapi.CallbackBeforeSetFriendRemarkResp{Remark: req.Remark}, nil + }) + RegisterCallback[cbapi.CallbackAfterSetFriendRemarkReq, cbapi.CallbackAfterSetFriendRemarkResp](s, cbapi.CallbackAfterSetFriendRemarkCommand, func(c *gin.Context, req *cbapi.CallbackAfterSetFriendRemarkReq) (*cbapi.CallbackAfterSetFriendRemarkResp, error) { + return &cbapi.CallbackAfterSetFriendRemarkResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeAddBlackReq, cbapi.CallbackBeforeAddBlackResp](s, cbapi.CallbackBeforeAddBlackCommand, func(c *gin.Context, req *cbapi.CallbackBeforeAddBlackReq) (*cbapi.CallbackBeforeAddBlackResp, error) { + return &cbapi.CallbackBeforeAddBlackResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterRemoveBlackReq, cbapi.CallbackAfterRemoveBlackResp](s, cbapi.CallbackAfterRemoveBlackCommand, func(c *gin.Context, req *cbapi.CallbackAfterRemoveBlackReq) (*cbapi.CallbackAfterRemoveBlackResp, error) { + return &cbapi.CallbackAfterRemoveBlackResp{}, nil + }) + + // Group + RegisterCallback[cbapi.CallbackBeforeCreateGroupReq, cbapi.CallbackBeforeCreateGroupResp](s, cbapi.CallbackBeforeCreateGroupCommand, func(c *gin.Context, req *cbapi.CallbackBeforeCreateGroupReq) (*cbapi.CallbackBeforeCreateGroupResp, error) { + return &cbapi.CallbackBeforeCreateGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterCreateGroupReq, cbapi.CallbackAfterCreateGroupResp](s, cbapi.CallbackAfterCreateGroupCommand, func(c *gin.Context, req *cbapi.CallbackAfterCreateGroupReq) (*cbapi.CallbackAfterCreateGroupResp, error) { + return &cbapi.CallbackAfterCreateGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeMembersJoinGroupReq, cbapi.CallbackBeforeMembersJoinGroupResp](s, cbapi.CallbackBeforeMembersJoinGroupCommand, func(c *gin.Context, req *cbapi.CallbackBeforeMembersJoinGroupReq) (*cbapi.CallbackBeforeMembersJoinGroupResp, error) { + return &cbapi.CallbackBeforeMembersJoinGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeMembersJoinGroupReq, cbapi.CallbackBeforeMembersJoinGroupResp](s, cbapi.CallbackBeforeJoinGroupCommand, func(c *gin.Context, req *cbapi.CallbackBeforeMembersJoinGroupReq) (*cbapi.CallbackBeforeMembersJoinGroupResp, error) { + return &cbapi.CallbackBeforeMembersJoinGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterJoinGroupReq, cbapi.CallbackAfterJoinGroupResp](s, cbapi.CallbackAfterJoinGroupCommand, func(c *gin.Context, req *cbapi.CallbackAfterJoinGroupReq) (*cbapi.CallbackAfterJoinGroupResp, error) { + return &cbapi.CallbackAfterJoinGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackQuitGroupReq, cbapi.CallbackQuitGroupResp](s, cbapi.CallbackAfterQuitGroupCommand, func(c *gin.Context, req *cbapi.CallbackQuitGroupReq) (*cbapi.CallbackQuitGroupResp, error) { + return &cbapi.CallbackQuitGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackKillGroupMemberReq, cbapi.CallbackKillGroupMemberResp](s, cbapi.CallbackAfterKickGroupCommand, func(c *gin.Context, req *cbapi.CallbackKillGroupMemberReq) (*cbapi.CallbackKillGroupMemberResp, error) { + return &cbapi.CallbackKillGroupMemberResp{}, nil + }) + RegisterCallback[cbapi.CallbackDisMissGroupReq, cbapi.CallbackDisMissGroupResp](s, cbapi.CallbackAfterDisMissGroupCommand, func(c *gin.Context, req *cbapi.CallbackDisMissGroupReq) (*cbapi.CallbackDisMissGroupResp, error) { + return &cbapi.CallbackDisMissGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackTransferGroupOwnerReq, cbapi.CallbackTransferGroupOwnerResp](s, cbapi.CallbackAfterTransferGroupOwnerCommand, func(c *gin.Context, req *cbapi.CallbackTransferGroupOwnerReq) (*cbapi.CallbackTransferGroupOwnerResp, error) { + return &cbapi.CallbackTransferGroupOwnerResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeInviteUserToGroupReq, cbapi.CallbackBeforeInviteUserToGroupResp](s, cbapi.CallbackBeforeInviteJoinGroupCommand, func(c *gin.Context, req *cbapi.CallbackBeforeInviteUserToGroupReq) (*cbapi.CallbackBeforeInviteUserToGroupResp, error) { + return &cbapi.CallbackBeforeInviteUserToGroupResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeSetGroupInfoReq, cbapi.CallbackBeforeSetGroupInfoResp](s, cbapi.CallbackBeforeSetGroupInfoCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSetGroupInfoReq) (*cbapi.CallbackBeforeSetGroupInfoResp, error) { + return &cbapi.CallbackBeforeSetGroupInfoResp{GroupID: req.GroupID, GroupName: req.GroupName, Notification: req.Notification, Introduction: req.Introduction, FaceURL: req.FaceURL}, nil + }) + RegisterCallback[cbapi.CallbackAfterSetGroupInfoReq, cbapi.CallbackAfterSetGroupInfoResp](s, cbapi.CallbackAfterSetGroupInfoCommand, func(c *gin.Context, req *cbapi.CallbackAfterSetGroupInfoReq) (*cbapi.CallbackAfterSetGroupInfoResp, error) { + return &cbapi.CallbackAfterSetGroupInfoResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeSetGroupInfoExReq, cbapi.CallbackBeforeSetGroupInfoExResp](s, cbapi.CallbackBeforeSetGroupInfoExCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSetGroupInfoExReq) (*cbapi.CallbackBeforeSetGroupInfoExResp, error) { + return &cbapi.CallbackBeforeSetGroupInfoExResp{GroupID: req.GroupID, GroupName: req.GroupName, Notification: req.Notification, Introduction: req.Introduction, FaceURL: req.FaceURL, Ex: req.Ex, NeedVerification: req.NeedVerification, LookMemberInfo: req.LookMemberInfo, ApplyMemberFriend: req.ApplyMemberFriend}, nil + }) + RegisterCallback[cbapi.CallbackAfterSetGroupInfoExReq, cbapi.CallbackAfterSetGroupInfoExResp](s, cbapi.CallbackAfterSetGroupInfoExCommand, func(c *gin.Context, req *cbapi.CallbackAfterSetGroupInfoExReq) (*cbapi.CallbackAfterSetGroupInfoExResp, error) { + return &cbapi.CallbackAfterSetGroupInfoExResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeSetGroupMemberInfoReq, cbapi.CallbackBeforeSetGroupMemberInfoResp](s, cbapi.CallbackBeforeSetGroupMemberInfoCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSetGroupMemberInfoReq) (*cbapi.CallbackBeforeSetGroupMemberInfoResp, error) { + return &cbapi.CallbackBeforeSetGroupMemberInfoResp{Nickname: req.Nickname, FaceURL: req.FaceURL, RoleLevel: req.RoleLevel, Ex: req.Ex}, nil + }) + RegisterCallback[cbapi.CallbackAfterSetGroupMemberInfoReq, cbapi.CallbackAfterSetGroupMemberInfoResp](s, cbapi.CallbackAfterSetGroupMemberInfoCommand, func(c *gin.Context, req *cbapi.CallbackAfterSetGroupMemberInfoReq) (*cbapi.CallbackAfterSetGroupMemberInfoResp, error) { + return &cbapi.CallbackAfterSetGroupMemberInfoResp{}, nil + }) + + // Message & Push & Revoke + RegisterCallback[cbapi.CallbackBeforeSendSingleMsgReq, cbapi.CallbackBeforeSendSingleMsgResp](s, cbapi.CallbackBeforeSendSingleMsgCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSendSingleMsgReq) (*cbapi.CallbackBeforeSendSingleMsgResp, error) { + return &cbapi.CallbackBeforeSendSingleMsgResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterSendSingleMsgReq, cbapi.CallbackAfterSendSingleMsgResp](s, cbapi.CallbackAfterSendSingleMsgCommand, func(c *gin.Context, req *cbapi.CallbackAfterSendSingleMsgReq) (*cbapi.CallbackAfterSendSingleMsgResp, error) { + return &cbapi.CallbackAfterSendSingleMsgResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforeSendGroupMsgReq, cbapi.CallbackBeforeSendGroupMsgResp](s, cbapi.CallbackBeforeSendGroupMsgCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSendGroupMsgReq) (*cbapi.CallbackBeforeSendGroupMsgResp, error) { + return &cbapi.CallbackBeforeSendGroupMsgResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterSendGroupMsgReq, cbapi.CallbackAfterSendGroupMsgResp](s, cbapi.CallbackAfterSendGroupMsgCommand, func(c *gin.Context, req *cbapi.CallbackAfterSendGroupMsgReq) (*cbapi.CallbackAfterSendGroupMsgResp, error) { + return &cbapi.CallbackAfterSendGroupMsgResp{}, nil + }) + RegisterCallback[cbapi.CallbackMsgModifyCommandReq, cbapi.CallbackMsgModifyCommandResp](s, cbapi.CallbackBeforeMsgModifyCommand, func(c *gin.Context, req *cbapi.CallbackMsgModifyCommandReq) (*cbapi.CallbackMsgModifyCommandResp, error) { + return &cbapi.CallbackMsgModifyCommandResp{}, nil + }) + RegisterCallback[cbapi.CallbackSingleMsgReadReq, cbapi.CallbackSingleMsgReadResp](s, cbapi.CallbackAfterSingleMsgReadCommand, func(c *gin.Context, req *cbapi.CallbackSingleMsgReadReq) (*cbapi.CallbackSingleMsgReadResp, error) { + return &cbapi.CallbackSingleMsgReadResp{}, nil + }) + RegisterCallback[cbapi.CallbackGroupMsgReadReq, cbapi.CallbackGroupMsgReadResp](s, cbapi.CallbackAfterGroupMsgReadCommand, func(c *gin.Context, req *cbapi.CallbackGroupMsgReadReq) (*cbapi.CallbackGroupMsgReadResp, error) { + return &cbapi.CallbackGroupMsgReadResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterMsgSaveDBReq, cbapi.CallbackAfterMsgSaveDBResp](s, cbapi.CallbackAfterMsgSaveDBCommand, func(c *gin.Context, req *cbapi.CallbackAfterMsgSaveDBReq) (*cbapi.CallbackAfterMsgSaveDBResp, error) { + return &cbapi.CallbackAfterMsgSaveDBResp{}, nil + }) + RegisterCallback[cbapi.CallbackAfterRevokeMsgReq, cbapi.CallbackAfterRevokeMsgResp](s, cbapi.CallbackAfterRevokeMsgCommand, func(c *gin.Context, req *cbapi.CallbackAfterRevokeMsgReq) (*cbapi.CallbackAfterRevokeMsgResp, error) { + return &cbapi.CallbackAfterRevokeMsgResp{}, nil + }) + RegisterCallback[cbapi.CallbackBeforePushReq, cbapi.CallbackBeforePushResp](s, cbapi.CallbackBeforeOfflinePushCommand, func(c *gin.Context, req *cbapi.CallbackBeforePushReq) (*cbapi.CallbackBeforePushResp, error) { + return &cbapi.CallbackBeforePushResp{UserIDs: req.UserIDList, OfflinePushInfo: req.OfflinePushInfo}, nil + }) + RegisterCallback[cbapi.CallbackBeforePushReq, cbapi.CallbackBeforePushResp](s, cbapi.CallbackBeforeOnlinePushCommand, func(c *gin.Context, req *cbapi.CallbackBeforePushReq) (*cbapi.CallbackBeforePushResp, error) { + return &cbapi.CallbackBeforePushResp{UserIDs: req.UserIDList}, nil + }) + RegisterCallback[cbapi.CallbackBeforeSuperGroupOnlinePushReq, cbapi.CallbackBeforeSuperGroupOnlinePushResp](s, cbapi.CallbackBeforeGroupOnlinePushCommand, func(c *gin.Context, req *cbapi.CallbackBeforeSuperGroupOnlinePushReq) (*cbapi.CallbackBeforeSuperGroupOnlinePushResp, error) { + return &cbapi.CallbackBeforeSuperGroupOnlinePushResp{}, nil + }) +} diff --git a/test/webhook/server.go b/test/webhook/server.go new file mode 100644 index 000000000..973254973 --- /dev/null +++ b/test/webhook/server.go @@ -0,0 +1,184 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package webhook + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "sync" + "time" + + "github.com/gin-gonic/gin" + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" +) + +func init() { + gin.SetMode(gin.TestMode) +} + +// RequestRecord stores information about a received webhook request. +type RequestRecord struct { + Timestamp time.Time + Command string + Body []byte + Headers http.Header +} + +// Server is a mock webhook HTTP server for OpenIM callbacks. +type Server struct { + Engine *gin.Engine + handlers map[string]gin.HandlerFunc + mu sync.RWMutex + httpServer *http.Server + recordHistory bool + history []RequestRecord + requestHook func(command string, reqBody []byte, c *gin.Context) +} + +// NewServer creates a new mock webhook server instance. +func NewServer() *Server { + engine := gin.New() + engine.Use(gin.Recovery()) + + s := &Server{ + Engine: engine, + handlers: make(map[string]gin.HandlerFunc), + } + + engine.POST("/callbackExample/:command", func(c *gin.Context) { + s.dispatch(c, c.Param("command")) + }) + engine.POST("/:command", func(c *gin.Context) { + s.dispatch(c, c.Param("command")) + }) + + s.RegisterDefaultHandlers() + return s +} + +// SetRecordHistory enables or disables recording incoming request history. +func (s *Server) SetRecordHistory(enabled bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.recordHistory = enabled +} + +// OnRequest sets a callback function invoked whenever any webhook request arrives. +func (s *Server) OnRequest(hook func(command string, reqBody []byte, c *gin.Context)) { + s.mu.Lock() + defer s.mu.Unlock() + s.requestHook = hook +} + +// GetHistory returns a copy of all recorded requests. +func (s *Server) GetHistory() []RequestRecord { + s.mu.RLock() + defer s.mu.RUnlock() + result := make([]RequestRecord, len(s.history)) + copy(result, s.history) + return result +} + +// ClearHistory clears all recorded request history. +func (s *Server) ClearHistory() { + s.mu.Lock() + defer s.mu.Unlock() + s.history = nil +} + +// Start runs the webhook HTTP server on the given address (e.g. ":10006"). +func (s *Server) Start(addr string) error { + s.httpServer = &http.Server{ + Addr: addr, + Handler: s.Engine, + } + return s.httpServer.ListenAndServe() +} + +// Shutdown gracefully stops the running HTTP server. +func (s *Server) Shutdown(ctx context.Context) error { + if s.httpServer == nil { + return nil + } + return s.httpServer.Shutdown(ctx) +} + +func (s *Server) dispatch(c *gin.Context, command string) { + s.mu.RLock() + h, ok := s.handlers[command] + s.mu.RUnlock() + if ok { + h(c) + return + } + c.JSON(http.StatusOK, cbapi.CommonCallbackResp{}) +} + +// ServeHTTP conforms to the http.Handler interface. +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + s.Engine.ServeHTTP(w, r) +} + +// RegisterCallback registers a typed handler for a callback command. +func RegisterCallback[Req any, Resp any](s *Server, command string, fn func(c *gin.Context, req *Req) (*Resp, error)) { + handler := func(c *gin.Context) { + body, err := io.ReadAll(c.Request.Body) + if err != nil { + c.String(http.StatusInternalServerError, err.Error()) + return + } + + s.mu.Lock() + if s.recordHistory { + s.history = append(s.history, RequestRecord{ + Timestamp: time.Now(), + Command: command, + Body: body, + Headers: c.Request.Header.Clone(), + }) + } + hook := s.requestHook + s.mu.Unlock() + + if hook != nil { + hook(command, body, c) + } + + var req Req + if len(body) > 0 { + if err := json.Unmarshal(body, &req); err != nil { + c.String(http.StatusBadRequest, fmt.Sprintf("unmarshal error: %v", err)) + return + } + } + resp, err := fn(c, &req) + if err != nil { + c.String(http.StatusInternalServerError, err.Error()) + return + } + if resp != nil { + c.JSON(http.StatusOK, resp) + } else { + c.JSON(http.StatusOK, cbapi.CommonCallbackResp{}) + } + } + + s.mu.Lock() + s.handlers[command] = handler + s.mu.Unlock() +} diff --git a/test/webhook/server_test.go b/test/webhook/server_test.go new file mode 100644 index 000000000..5235a91a9 --- /dev/null +++ b/test/webhook/server_test.go @@ -0,0 +1,122 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package webhook + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" + sdkws "github.com/openimsdk/protocol/sdkws" + "github.com/stretchr/testify/require" +) + +func TestWebhookServer_DefaultEndpoints(t *testing.T) { + t.Parallel() + srv := NewServer() + + tests := []struct { + name string + url string + payload any + expectRes int + }{ + { + name: "CallbackBeforeUserRegister with prefix", + url: "/callbackExample/" + cbapi.CallbackBeforeUserRegisterCommand, + payload: &cbapi.CallbackBeforeUserRegisterReq{ + CallbackCommand: cbapi.CallbackCommand(cbapi.CallbackBeforeUserRegisterCommand), + Users: []*sdkws.UserInfo{{UserID: "u100", Nickname: "Alice"}}, + }, + expectRes: http.StatusOK, + }, + { + name: "CallbackAfterMsgSaveDB without prefix", + url: "/" + cbapi.CallbackAfterMsgSaveDBCommand, + payload: &cbapi.CallbackAfterMsgSaveDBReq{ + CommonCallbackReq: cbapi.CommonCallbackReq{ + SendID: "u100", + CallbackCommand: cbapi.CallbackAfterMsgSaveDBCommand, + }, + RecvID: "u200", + }, + expectRes: http.StatusOK, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + body, err := json.Marshal(tt.payload) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, tt.url, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + srv.ServeHTTP(w, req) + require.Equal(t, tt.expectRes, w.Code) + }) + } +} + +func TestWebhookServer_CustomHandlerOverride(t *testing.T) { + t.Parallel() + srv := NewServer() + + // Override BeforeMsgModify to mutate content + filteredContent := `{"text":"***"}` + RegisterCallback( + srv, + cbapi.CallbackBeforeMsgModifyCommand, + func(c *gin.Context, req *cbapi.CallbackMsgModifyCommandReq) (*cbapi.CallbackMsgModifyCommandResp, error) { + return &cbapi.CallbackMsgModifyCommandResp{ + CommonCallbackResp: cbapi.CommonCallbackResp{ + ErrCode: 0, + ErrMsg: "filtered", + }, + Content: &filteredContent, + }, nil + }, + ) + + reqPayload := cbapi.CallbackMsgModifyCommandReq{ + CommonCallbackReq: cbapi.CommonCallbackReq{ + SendID: "u1", + CallbackCommand: cbapi.CallbackBeforeMsgModifyCommand, + }, + } + body, err := json.Marshal(reqPayload) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, "/"+cbapi.CallbackBeforeMsgModifyCommand, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + srv.ServeHTTP(w, req) + require.Equal(t, http.StatusOK, w.Code) + + var resp cbapi.CallbackMsgModifyCommandResp + err = json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + require.NotNil(t, resp.Content) + require.Equal(t, `{"text":"***"}`, *resp.Content) + require.Equal(t, "filtered", resp.ErrMsg) +}