feat: Msg filter (#2703)

* feat: msg filter

* feat: msg filter

* feat: msg filter

config/webhooks.yml

@@ -3,6 +3,16 @@ beforeSendSingleMsg:
   enable: false
   timeout: 5
   failedContinue: true
+  # Only the contentType in allowedTypes will send the callback.
+  # Supports two formats: a single type or a range. The range is defined by the lower and upper bounds connected with a hyphen ("-").
+  # e.g. allowedTypes: [1, 100, 200-500, 600-700] means that only contentType within the range
+  # {1, 100} ∪ [200, 500] ∪ [600, 700] will be allowed through the filter.
+  # If not set, all contentType messages will through this filter.
+  allowedTypes: []
+  # Only the contentType not in deniedTypes will send the callback.
+  # Supports two formats, same as allowedTypes.
+  # If not set, all contentType messages will through this filter.
+  deniedTypes: []
 beforeUpdateUserInfoEx:
   enable:  false
   timeout: 5
@@ -16,17 +26,29 @@ afterSendSingleMsg:
   # Only the senID/recvID specified in attentionIds will send the callback
   # if not set, all user messages will be callback
   attentionIds: []
+  # See beforeSendSingleMsg comment.
+  allowedTypes: []
+  deniedTypes: []
 beforeSendGroupMsg:
   enable: false
   timeout: 5
   failedContinue: true
+  # See beforeSendSingleMsg comment.
+  allowedTypes: []
+  deniedTypes: []
 beforeMsgModify:
   enable: false
   timeout: 5
   failedContinue: true
+  # See beforeSendSingleMsg comment.
+  allowedTypes: []
+  deniedTypes: []
 afterSendGroupMsg:
   enable: false
   timeout: 5
+  # See beforeSendSingleMsg comment.
+  allowedTypes: []
+  deniedTypes: []
 afterUserOnline:
   enable: false
   timeout: 5

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/openimsdk/protocol v0.0.72
-	github.com/openimsdk/tools v0.0.50-alpha.15
+	github.com/openimsdk/tools v0.0.50-alpha.16
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0

go.sum

@@ -321,8 +321,8 @@ github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCF
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
 github.com/openimsdk/protocol v0.0.72 h1:K+vslwaR7lDXyBzb07UuEQITaqsgighz7NyXVIWsu6A=
 github.com/openimsdk/protocol v0.0.72/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
-github.com/openimsdk/tools v0.0.50-alpha.15 h1:HV9aKZ4vvCZCGG4wFDsgUONkkdJeCcrFNn3BT52nUVQ=
-github.com/openimsdk/tools v0.0.50-alpha.15/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4=
+github.com/openimsdk/tools v0.0.50-alpha.16 h1:bC1AQvJMuOHtZm8LZRvN8L5mH1Ws2VYdL+TLTs1iGSc=
+github.com/openimsdk/tools v0.0.50-alpha.16/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

internal/rpc/msg/callback.go

@@ -67,6 +67,9 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf
 		if msg.MsgData.ContentType == constant.Typing {
 			return nil
 		}
+		if !filterBeforeMsg(msg, before) {
+			return nil
+		}
 		cbReq := &cbapi.CallbackBeforeSendSingleMsgReq{
 			CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackBeforeSendSingleMsgCommand),
 			RecvID:            msg.MsgData.RecvID,
@@ -84,9 +87,7 @@ func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config
 	if msg.MsgData.ContentType == constant.Typing {
 		return
 	}
-	// According to the attentionIds configuration, only some users are sent
-	attentionIds := after.AttentionIds
-	if attentionIds != nil && !datautil.Contain(msg.MsgData.RecvID, attentionIds...) && !datautil.Contain(msg.MsgData.SendID, attentionIds...) {
+	if !filterAfterMsg(msg, after) {
 		return
 	}
 	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
@@ -98,6 +99,9 @@ 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) {
+			return nil
+		}
 		if msg.MsgData.ContentType == constant.Typing {
 			return nil
 		}
@@ -117,6 +121,9 @@ func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.
 	if msg.MsgData.ContentType == constant.Typing {
 		return
 	}
+	if !filterAfterMsg(msg, after) {
+		return
+	}
 	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
 		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
 		GroupID:           msg.MsgData.GroupID,
@@ -129,6 +136,9 @@ func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.B
 		if msg.MsgData.ContentType != constant.Text {
 			return nil
 		}
+		if !filterBeforeMsg(msg, before) {
+			return nil
+		}
 		cbReq := &cbapi.CallbackMsgModifyCommandReq{
 			CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackBeforeMsgModifyCommand),
 		}

internal/rpc/msg/filter.go

@@ -0,0 +1,67 @@
+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/common/config/config.go

@@ -345,15 +345,19 @@ type Redis struct {
 }
 
 type BeforeConfig struct {
-	Enable         bool `mapstructure:"enable"`
-	Timeout        int  `mapstructure:"timeout"`
-	FailedContinue bool `mapstructure:"failedContinue"`
+	Enable         bool     `mapstructure:"enable"`
+	Timeout        int      `mapstructure:"timeout"`
+	FailedContinue bool     `mapstructure:"failedContinue"`
+	AllowedTypes   []string `mapstructure:"allowedTypes"`
+	DeniedTypes    []string `mapstructure:"deniedTypes"`
 }
 
 type AfterConfig struct {
 	Enable       bool     `mapstructure:"enable"`
 	Timeout      int      `mapstructure:"timeout"`
 	AttentionIds []string `mapstructure:"attentionIds"`
+	AllowedTypes []string `mapstructure:"allowedTypes"`
+	DeniedTypes  []string `mapstructure:"deniedTypes"`
 }
 
 type Share struct {