blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-09-04 22:53:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: afterMsgSaveDB webhook (#3788)

Browse Source 
* 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
This commit is contained in:
dsx137 2026-08-21 19:36:11 +08:00 committed by GitHub
parent 9928d3fba3
commit 49ec272f88
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 1287 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -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

6
config/webhooks.yml
View File

@ -49,6 +49,12 @@ afterSendGroupMsg:
# See beforeSendSingleMsg comment.
allowedTypes: []
deniedTypes: []
afterMsgSaveDB:
enable: false
timeout: 5
attentionIds: []
allowedTypes: []
deniedTypes: []
afterUserOnline:
enable: false
timeout: 5

73
internal/msgtransfer/callback.go Normal file
View File

@ -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)
}

102
internal/msgtransfer/callback_test.go Normal file
View File

@ -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)
}

2
internal/msgtransfer/init.go
View File

@ -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
}

10
internal/msgtransfer/online_msg_to_mongo_handler.go
View File

@ -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 {

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

@ -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{

67
internal/rpc/msg/filter.go
View File

@ -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
}

1
pkg/callbackstruct/constant.go
View File

@ -62,4 +62,5 @@ const (
CallbackBeforeMembersJoinGroupCommand = "callbackBeforeMembersJoinGroupCommand"
CallbackBeforeSetGroupMemberInfoCommand = "callbackBeforeSetGroupMemberInfoCommand"
CallbackAfterSetGroupMemberInfoCommand = "callbackAfterSetGroupMemberInfoCommand"
CallbackAfterMsgSaveDBCommand = "callbackAfterMsgSaveDBCommand"
)

10
pkg/callbackstruct/message.go
View File

@ -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
}

1
pkg/common/config/config.go
View File

@ -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"`

3
pkg/common/config/load_config_test.go
View File

@ -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) {

62
pkg/common/webhook/content_type.go Normal file
View File

@ -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
}

171
pkg/common/webhook/content_type_test.go Normal file
View File

@ -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)
}
})
}
}

3
pkg/common/webhook/http_client.go
View File

@ -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 {

45
pkg/common/webhook/http_client_test.go
View File

@ -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)
}
}

209
test/webhook/callbacks_test.go Normal file
View File

@ -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",
})
})
}
}

103
test/webhook/cmd/main.go Normal file
View File

@ -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.")
}

177
test/webhook/default_handlers.go Normal file
View File

@ -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
})
}

184
test/webhook/server.go Normal file
View File

@ -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()
}

122
test/webhook/server_test.go Normal file
View File

@ -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)
}