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

.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

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

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

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

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
 	}

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 {

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{

internal/rpc/msg/filter.go

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

pkg/callbackstruct/constant.go

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

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

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

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

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

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

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

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