Merge 5556e89fa3d46a521247c755de0befcd839c17d9 into 1baf9a8e0f11cc94313eec2165e273482d1bdb0d

config/redis.yml

@@ -1,14 +1,7 @@
-address: [localhost:16379]
-username:
+address: [ localhost:16379 ]
+username: 
 password: openIM123
-# redis Mode, including "standalone","cluster","sentinel"
-redisMode: "standalone"
+clusterMode: false
 db: 0
 maxRetry: 10
 poolSize: 100
-# Sentinel configuration (only used when redisMode is "sentinel")
-sentinelMode:
-  masterName: "redis-master"
-  sentinelsAddrs: ["127.0.0.1:26379", "127.0.0.1:26380", "127.0.0.1:26381"]
-  routeByLatency: true
-  routeRandomly: true

config/webhooks.yml

@@ -16,7 +16,7 @@ afterUpdateUserInfoEx:
 afterSendSingleMsg:
   enable: false
   timeout: 5
-  # Only the recvIDs specified in attentionIds will send the callback
+  # Only the recvID specified in attentionIds will send the callback
   # if not set, all user messages will be callback
   attentionIds: []
   # See beforeSendSingleMsg comment.
@@ -36,7 +36,7 @@ beforeMsgModify:
 afterSendGroupMsg:
   enable: false
   timeout: 5
-  # Only the GroupIDs specified in attentionIds will send the callback
+  # Only the recvID specified in attentionIds will send the callback
   # if not set, all user messages will be callback
   attentionIds: []
   # See beforeSendSingleMsg comment.

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.73-alpha.12
-	github.com/openimsdk/tools v0.0.50-alpha.92
+	github.com/openimsdk/tools v0.0.50-alpha.85
 	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

@@ -349,8 +349,8 @@ github.com/openimsdk/gomake v0.0.15-alpha.5 h1:eEZCEHm+NsmcO3onXZPIUbGFCYPYbsX5b
 github.com/openimsdk/gomake v0.0.15-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
 github.com/openimsdk/protocol v0.0.73-alpha.12 h1:2NYawXeHChYUeSme6QJ9pOLh+Empce2WmwEtbP4JvKk=
 github.com/openimsdk/protocol v0.0.73-alpha.12/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
-github.com/openimsdk/tools v0.0.50-alpha.92 h1:hWfykMhmi7EQEiwgQccJqbgggIuhun/PrVkBnjmj9Ec=
-github.com/openimsdk/tools v0.0.50-alpha.92/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
+github.com/openimsdk/tools v0.0.50-alpha.85 h1:OqTUYx6r7Zp/eH8FKB08XeNjPV405TUIG9QT6QQ+F+s=
+github.com/openimsdk/tools v0.0.50-alpha.85/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
 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/api/router.go

@@ -28,7 +28,6 @@ import (
 	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mw"
-	"github.com/openimsdk/tools/mw/api"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
@@ -97,7 +96,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
-	r.Use(api.GinLogger(), prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
+	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
 		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)), setGinIsAdmin(cfg.Share.IMAdminUserID))
 
 	u := NewUserApi(user.NewUserClient(userConn), client, cfg.Discovery.RpcService)

internal/msgtransfer/callback.go

@@ -55,11 +55,9 @@ func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendSingleMsg(ctx conte
 	if msg.ContentType == constant.Typing {
 		return
 	}
-
 	if !filterAfterMsg(msg, after) {
 		return
 	}
-
 	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
 		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
 		RecvID:            msg.RecvID,
@@ -71,11 +69,9 @@ func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendGroupMsg(ctx contex
 	if msg.ContentType == constant.Typing {
 		return
 	}
-
 	if !filterAfterMsg(msg, after) {
 		return
 	}
-
 	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
 		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
 		GroupID:           msg.GroupID,
@@ -102,11 +98,7 @@ func filterAfterMsg(msg *sdkws.MsgData, after *config.AfterConfig) bool {
 
 func filterMsg(msg *sdkws.MsgData, attentionIds []string, deniedTypes []int32) bool {
 	// According to the attentionIds configuration, only some users are sent
-	if len(attentionIds) != 0 && msg.ContentType == constant.SingleChatType && !datautil.Contain(msg.RecvID, attentionIds...) {
-		return false
-	}
-
-	if len(attentionIds) != 0 && msg.ContentType == constant.ReadGroupChatType && !datautil.Contain(msg.GroupID, attentionIds...) {
+	if len(attentionIds) != 0 && !datautil.Contain(msg.RecvID, attentionIds...) {
 		return false
 	}
 

internal/rpc/msg/callback.go

@@ -16,10 +16,13 @@ package msg
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/utils/stringutil"
 
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -86,20 +89,19 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf
 	})
 }
 
-// Move to msgtransfer
-// func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
-// 	if msg.MsgData.ContentType == constant.Typing {
-// 		return
-// 	}
-// 	if !filterAfterMsg(msg, after) {
-// 		return
-// 	}
-// 	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
-// 		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
-// 		RecvID:            msg.MsgData.RecvID,
-// 	}
-// 	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
-// }
+func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
+	if msg.MsgData.ContentType == constant.Typing {
+		return
+	}
+	if !filterAfterMsg(msg, after) {
+		return
+	}
+	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
+		RecvID:            msg.MsgData.RecvID,
+	}
+	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
+}
 
 func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@@ -121,21 +123,20 @@ func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *confi
 	})
 }
 
-// Move to msgtransfer
-// func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
-// 	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,
-// 	}
+func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) {
+	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,
+	}
 
-// 	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
-// }
+	m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData))
+}
 
 func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq, beforeMsgData **sdkws.MsgData) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@@ -204,14 +205,14 @@ func (m *msgServer) webhookAfterRevokeMsg(ctx context.Context, after *config.Aft
 	m.webhookClient.AsyncPost(ctx, callbackReq.GetCallbackCommand(), callbackReq, &cbapi.CallbackAfterRevokeMsgResp{}, after)
 }
 
-// func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {
-// 	keyMsgData := apistruct.KeyMsgData{
-// 		SendID:  msg.SendID,
-// 		RecvID:  msg.RecvID,
-// 		GroupID: msg.GroupID,
-// 	}
+func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string {
+	keyMsgData := apistruct.KeyMsgData{
+		SendID:  msg.SendID,
+		RecvID:  msg.RecvID,
+		GroupID: msg.GroupID,
+	}
 
-// 	return map[string]string{
-// 		webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)),
-// 	}
-// }
+	return map[string]string{
+		webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)),
+	}
+}

internal/tools/cron/cron_task.go

@@ -49,7 +49,6 @@ func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegis
 		return err
 	}
 
-	var locker Locker
 	if conf.Discovery.Enable == config.ETCD {
 		cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), []string{
 			conf.CronTask.GetConfigFileName(),
@@ -57,14 +56,11 @@ func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegis
 			conf.Discovery.GetConfigFileName(),
 		})
 		cm.Watch(ctx)
-		locker, err = NewEtcdLocker(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient())
-		if err != nil {
-			return err
-		}
 	}
 
-	if locker == nil {
-		locker = emptyLocker{}
+	locker, err := NewEtcdLocker(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient())
+	if err != nil {
+		return err
 	}
 
 	srv := &cronServer{
@@ -96,16 +92,6 @@ func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegis
 	return nil
 }
 
-type Locker interface {
-	ExecuteWithLock(ctx context.Context, taskName string, task func())
-}
-
-type emptyLocker struct{}
-
-func (emptyLocker) ExecuteWithLock(ctx context.Context, taskName string, task func()) {
-	task()
-}
-
 type cronServer struct {
 	ctx                context.Context
 	config             *Config
@@ -113,7 +99,7 @@ type cronServer struct {
 	msgClient          msg.MsgClient
 	conversationClient pbconversation.ConversationClient
 	thirdClient        third.ThirdClient
-	locker             Locker
+	locker             *EtcdLocker
 }
 
 func (c *cronServer) registerClearS3() error {

pkg/common/config/config.go

@@ -323,22 +323,14 @@ type RPC struct {
 }
 
 type Redis struct {
-	Disable      bool     `yaml:"-"`
-	Address      []string `yaml:"address"`
-	Username     string   `yaml:"username"`
-	Password     string   `yaml:"password"`
-	RedisMode    string   `yaml:"redisMode"`
-	DB           int      `yaml:"db"`
-	MaxRetry     int      `yaml:"maxRetry"`
-	PoolSize     int      `yaml:"poolSize"`
-	SentinelMode Sentinel `yaml:"sentinelMode"`
-}
-
-type Sentinel struct {
-	MasterName     string   `yaml:"masterName"`
-	SentinelAddrs  []string `yaml:"sentinelsAddrs"`
-	RouteByLatency bool     `yaml:"routeByLatency"`
-	RouteRandomly  bool     `yaml:"routeRandomly"`
+	Disable     bool     `yaml:"-"`
+	Address     []string `yaml:"address"`
+	Username    string   `yaml:"username"`
+	Password    string   `yaml:"password"`
+	ClusterMode bool     `yaml:"clusterMode"`
+	DB          int      `yaml:"db"`
+	MaxRetry    int      `yaml:"maxRetry"`
+	PoolSize    int      `yaml:"poolSize"`
 }
 
 type BeforeConfig struct {
@@ -495,19 +487,13 @@ func (m *Mongo) Build() *mongoutil.Config {
 
 func (r *Redis) Build() *redisutil.Config {
 	return &redisutil.Config{
-		RedisMode: r.RedisMode,
-		Address:   r.Address,
-		Username:  r.Username,
-		Password:  r.Password,
-		DB:        r.DB,
-		MaxRetry:  r.MaxRetry,
-		PoolSize:  r.PoolSize,
-		Sentinel: &redisutil.Sentinel{
-			MasterName:     r.SentinelMode.MasterName,
-			SentinelAddrs:  r.SentinelMode.SentinelAddrs,
-			RouteByLatency: r.SentinelMode.RouteByLatency,
-			RouteRandomly:  r.SentinelMode.RouteRandomly,
-		},
+		ClusterMode: r.ClusterMode,
+		Address:     r.Address,
+		Username:    r.Username,
+		Password:    r.Password,
+		DB:          r.DB,
+		MaxRetry:    r.MaxRetry,
+		PoolSize:    r.PoolSize,
 	}
 }
 

pkg/common/discovery/discoveryregister.go

@@ -35,7 +35,7 @@ func NewDiscoveryRegister(discovery *config.Discovery, watchNames []string) (dis
 		return standalone.GetSvcDiscoveryRegistry(), nil
 	}
 	if runtimeenv.RuntimeEnvironment() == config.KUBERNETES {
-		return kubernetes.NewConnManager(discovery.Kubernetes.Namespace, nil,
+		return kubernetes.NewKubernetesConnManager(discovery.Kubernetes.Namespace,
 			grpc.WithDefaultCallOptions(
 				grpc.MaxCallSendMsgSize(1024*1024*20),
 			),

pkg/common/storage/cache/redis/online_test.go

@@ -26,7 +26,7 @@ func TestName111111(t *testing.T) {
 			"172.16.8.124:7005",
 			"172.16.8.124:7006",
 		},
-		RedisMode: "cluster",
+		ClusterMode: true,
 		Password:    "passwd123",
 		//Address:  []string{"localhost:16379"},
 		//Password: "openIM123",

tools/seq/internal/seq.go

@@ -28,8 +28,6 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
-const StructTagName = "yaml"
-
 const (
 	MaxSeq                 = "MAX_SEQ:"
 	MinSeq                 = "MIN_SEQ:"
@@ -56,14 +54,13 @@ func readConfig[T any](dir string, name string) (*T, error) {
 	if err := v.ReadInConfig(); err != nil {
 		return nil, err
 	}
-
+	fn := func(config *mapstructure.DecoderConfig) {
+		config.TagName = "mapstructure"
+	}
 	var conf T
-	if err := v.Unmarshal(&conf, func(config *mapstructure.DecoderConfig) {
-		config.TagName = StructTagName
-	}); err != nil {
+	if err := v.Unmarshal(&conf, fn); err != nil {
 		return nil, err
 	}
-
 	return &conf, nil
 }
 
@@ -72,7 +69,6 @@ func Main(conf string, del time.Duration) error {
 	if err != nil {
 		return err
 	}
-	
 	mongodbConfig, err := readConfig[config.Mongo](conf, config.MongodbConfigFileName)
 	if err != nil {
 		return err