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..607c5ba13 100644 --- a/pkg/common/webhook/http_client.go +++ b/pkg/common/webhook/http_client.go @@ -62,7 +62,13 @@ func (c *Client) SyncPost(ctx context.Context, command string, req callbackstruc func (c *Client) AsyncPost(ctx context.Context, command string, req callbackstruct.CallbackReq, resp callbackstruct.CallbackResp, after *config.AfterConfig) { if after.Enable { - c.queue.Push(func() { c.post(ctx, command, req, resp, after.Timeout) }) + if err := c.queue.NotWaitPush(func() { + if err := c.post(ctx, command, req, resp, after.Timeout); err != nil { + log.ZError(ctx, "async webhook post failed", err, "url", c.url, "command", command) + } + }); err != nil { + log.ZError(ctx, "async webhook enqueue failed", err, "url", c.url, "command", command) + } } } 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) + } +}