Merge 5556e89fa3d46a521247c755de0befcd839c17d9 into 6912ad6f332e42b4b729783692e5971fdf00e7da

feat: add api logger (#3427 )
* pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * upgrading pkg tools * fix * fix * optimize log output * feat: support GetLastMessage * feat: support GetLastMessage * feat: s3 switch * feat: s3 switch * fix: GetUsersOnline * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: seq conversion failed without exiting * fix: DeleteDoc crash * fix: fill send time * fix: fill send time * fix: crash caused by withdrawing messages from users who have left the group * fix: user msg timestamp * seq read config * seq read config * fix: the source message of the reference is withdrawn, and the referenced message is deleted * feat: optimize the default notification.yml * fix: shouldPushOffline * fix: the sorting is wrong after canceling the administrator in group settings * feat: Sending messages supports returning fields modified by webhook * feat: Sending messages supports returning fields modified by webhook * feat: Sending messages supports returning fields modified by webhook * fix: oss specifies content-type when uploading * fix: the version number contains a line break * fix: the version number contains a line break * feat: GetConversationsHasReadAndMaxSeq support pinned * feat: GetConversationsHasReadAndMaxSeq support pinned * feat: GetConversationsHasReadAndMaxSeq support pinned * fix: transferring the group owner to a muted member, incremental version error * feat: unified conversion code * feat: update gomake * fix: in standalone mode, the user online status is wrong * fix: add permission check * fix: add permission check * fix: add rpc interface permission check * fix: CreateGroupChatConversations * feat: optimize friend and group applications * feat: optimize friend and group applications * feat: optimize friend and group applications * feat: optimize friend and group applications * fix: optimize friend and group applications * feat: add api logger
2026-07-08 21:31:13 +08:00 · 2025-06-20 22:27:44 +08:00 · 2025-06-19 03:51:35 +00:00 · 2025-06-18 06:32:29 +00:00 · 2025-06-18 06:31:09 +00:00
12 changed files with 121 additions and 74 deletions
--- a/config/redis.yml
+++ b/config/redis.yml
@ -1,7 +1,14 @@
-address: [ localhost:16379 ]
-username: 
+address: [localhost:16379]
+username:
 password: openIM123
-clusterMode: false
+# redis Mode, including "standalone","cluster","sentinel"
+redisMode: "standalone"
 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
--- a/config/webhooks.yml
+++ b/config/webhooks.yml
@ -16,7 +16,7 @@ afterUpdateUserInfoEx:
 afterSendSingleMsg:
  enable: false
  timeout: 5
-  # Only the recvID specified in attentionIds will send the callback
+  # Only the recvIDs 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 recvID specified in attentionIds will send the callback
+  # Only the GroupIDs specified in attentionIds will send the callback
  # if not set, all user messages will be callback
  attentionIds: []
  # See beforeSendSingleMsg comment.
--- a/go.mod
+++ b/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.85
+	github.com/openimsdk/tools v0.0.50-alpha.92
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0
--- a/go.sum
+++ b/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.85 h1:OqTUYx6r7Zp/eH8FKB08XeNjPV405TUIG9QT6QQ+F+s=
-github.com/openimsdk/tools v0.0.50-alpha.85/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
+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/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=
--- a/internal/api/router.go
+++ b/internal/api/router.go
@ -28,6 +28,7 @@ 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"
 )

@ -96,7 +97,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
-	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
+	r.Use(api.GinLogger(), 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)
--- a/internal/msgtransfer/callback.go
+++ b/internal/msgtransfer/callback.go
@ -55,9 +55,11 @@ 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,
@ -69,9 +71,11 @@ 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,
@ -98,7 +102,11 @@ 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 && !datautil.Contain(msg.RecvID, attentionIds...) {
+	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...) {
 		return false
 	}

--- a/internal/rpc/msg/callback.go
+++ b/internal/rpc/msg/callback.go
@ -16,13 +16,10 @@ 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"
@ -89,19 +86,20 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf
 	})
 }

-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))
-}
+// 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) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
@ -123,20 +121,21 @@ func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *confi
 	})
 }

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

-	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 {
@ -205,14 +204,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)),
+// 	}
+// }
--- a/internal/tools/cron/cron_task.go
+++ b/internal/tools/cron/cron_task.go
@ -49,6 +49,7 @@ 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(),
@ -56,11 +57,14 @@ 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
+		}
 	}

-	locker, err := NewEtcdLocker(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient())
-	if err != nil {
-		return err
+	if locker == nil {
+		locker = emptyLocker{}
 	}

 	srv := &cronServer{
@ -92,6 +96,16 @@ 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
@ -99,7 +113,7 @@ type cronServer struct {
 	msgClient          msg.MsgClient
 	conversationClient pbconversation.ConversationClient
 	thirdClient        third.ThirdClient
-	locker             *EtcdLocker
+	locker             Locker
 }

 func (c *cronServer) registerClearS3() error {
--- a/pkg/common/config/config.go
+++ b/pkg/common/config/config.go
@ -323,14 +323,22 @@ type RPC struct {
 }

 type Redis struct {
-	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"`
+	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"`
 }

 type BeforeConfig struct {
@ -487,13 +495,19 @@ func (m *Mongo) Build() *mongoutil.Config {

 func (r *Redis) Build() *redisutil.Config {
 	return &redisutil.Config{
-		ClusterMode: r.ClusterMode,
-		Address:     r.Address,
-		Username:    r.Username,
-		Password:    r.Password,
-		DB:          r.DB,
-		MaxRetry:    r.MaxRetry,
-		PoolSize:    r.PoolSize,
+		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,
+		},
 	}
 }

--- a/pkg/common/discovery/discoveryregister.go
+++ b/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.NewKubernetesConnManager(discovery.Kubernetes.Namespace,
+		return kubernetes.NewConnManager(discovery.Kubernetes.Namespace, nil,
 			grpc.WithDefaultCallOptions(
 				grpc.MaxCallSendMsgSize(1024*1024*20),
 			),
--- a/pkg/common/storage/cache/redis/online_test.go
+++ b/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",
 		},
-		ClusterMode: true,
+		RedisMode: "cluster",
 		Password:    "passwd123",
 		//Address:  []string{"localhost:16379"},
 		//Password: "openIM123",
--- a/tools/seq/internal/seq.go
+++ b/tools/seq/internal/seq.go
@ -28,6 +28,8 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/options"
 )

+const StructTagName = "yaml"
+
 const (
 	MaxSeq                 = "MAX_SEQ:"
 	MinSeq                 = "MIN_SEQ:"
@ -54,13 +56,14 @@ 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, fn); err != nil {
+	if err := v.Unmarshal(&conf, func(config *mapstructure.DecoderConfig) {
+		config.TagName = StructTagName
+	}); err != nil {
 		return nil, err
 	}
+
 	return &conf, nil
 }

@ -69,6 +72,7 @@ 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
Author	SHA1	Message	Date
OpenIM-Gordon	95b49391e7	Merge 5556e89fa3d46a521247c755de0befcd839c17d9 into 6912ad6f332e42b4b729783692e5971fdf00e7da	2025-06-20 22:27:44 +08:00
chao	6912ad6f33	feat: add api logger (#3427 ) * pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * upgrading pkg tools * fix * fix * optimize log output * feat: support GetLastMessage * feat: support GetLastMessage * feat: s3 switch * feat: s3 switch * fix: GetUsersOnline * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: seq conversion failed without exiting * fix: DeleteDoc crash * fix: fill send time * fix: fill send time * fix: crash caused by withdrawing messages from users who have left the group * fix: user msg timestamp * seq read config * seq read config * fix: the source message of the reference is withdrawn, and the referenced message is deleted * feat: optimize the default notification.yml * fix: shouldPushOffline * fix: the sorting is wrong after canceling the administrator in group settings * feat: Sending messages supports returning fields modified by webhook * feat: Sending messages supports returning fields modified by webhook * feat: Sending messages supports returning fields modified by webhook * fix: oss specifies content-type when uploading * fix: the version number contains a line break * fix: the version number contains a line break * feat: GetConversationsHasReadAndMaxSeq support pinned * feat: GetConversationsHasReadAndMaxSeq support pinned * feat: GetConversationsHasReadAndMaxSeq support pinned * fix: transferring the group owner to a muted member, incremental version error * feat: unified conversion code * feat: update gomake * fix: in standalone mode, the user online status is wrong * fix: add permission check * fix: add permission check * fix: add rpc interface permission check * fix: CreateGroupChatConversations * feat: optimize friend and group applications * feat: optimize friend and group applications * feat: optimize friend and group applications * feat: optimize friend and group applications * fix: optimize friend and group applications * feat: add api logger	2025-06-19 03:51:35 +00:00
Monet Lee	53bf8acc21	fix: solve webhook incorrect attentionID references. (#3411 )	2025-06-18 06:32:29 +00:00
Monet Lee	8f7b02979d	feat: support redis sentinel. (#3423 ) * feat: support redis sentinel. * update docker compose contents. * update config contents. * revert content. * supoort redisMode. * update config. * remvove print.	2025-06-18 06:31:09 +00:00