Merge remote-tracking branch 'origin/main' into pre-release-v3.8.4

# Conflicts: # .env # .github/workflows/docker-build-and-release-services-images.yml # .github/workflows/merge-from-milestone.yml # .github/workflows/update-version-file-on-release.yml # README.md # README_zh_CN.md # cmd/main.go # config/discovery.yml # config/notification.yml # config/openim-api.yml # config/openim-msggateway.yml # config/openim-msgtransfer.yml # config/openim-push.yml # config/openim-rpc-auth.yml # config/openim-rpc-conversation.yml # config/openim-rpc-friend.yml # config/openim-rpc-group.yml # config/openim-rpc-msg.yml # config/openim-rpc-third.yml # config/openim-rpc-user.yml # config/share.yml # config/webhooks.yml # deployments/templates/config.yaml # docker-compose.yml # go.mod # go.sum # internal/api/init.go # internal/api/jssdk/tools.go # internal/api/msg.go # internal/api/prometheus_discovery.go # internal/api/router.go # internal/msggateway/init.go # internal/msggateway/ws_server.go # internal/msgtransfer/init.go # internal/msgtransfer/online_history_msg_handler.go # internal/msgtransfer/online_msg_to_mongo_handler.go # internal/push/push.go # internal/rpc/auth/auth.go # internal/rpc/conversation/conversation.go # internal/rpc/group/group.go # internal/rpc/msg/callback.go # internal/rpc/msg/server.go # internal/rpc/relation/friend.go # internal/rpc/relation/notification.go # internal/rpc/third/third.go # internal/rpc/user/user.go # internal/tools/cron/cron_task.go # magefile.go # pkg/common/cmd/api.go # pkg/common/cmd/msg_transfer.go # pkg/common/config/config.go # pkg/common/discovery/discoveryregister.go # pkg/common/prommetrics/prommetrics.go # pkg/common/startrpc/start.go # pkg/common/storage/database/mgo/cache.go # pkg/common/storage/database/mgo/msg_test.go # pkg/rpcli/auth.go # pkg/rpcli/tool.go # pkg/rpcli/user.go # test/stress-test-v2/main.go # test/stress-test/main.go # tools/seq/internal/seq.go # version/version
2026-05-09 11:35:59 +08:00 · 2025-07-29 17:29:08 +08:00
parent b1837434df 55ebedc452
commit 08c7461ade
268 changed files with 2616 additions and 9031 deletions
@@ -39,7 +39,7 @@ type Config struct {
 	Index      conf.Index
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, service grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, service grpc.ServiceRegistrar) error {
 	apiPort, err := datautil.GetElemByIndex(config.API.Api.Ports, int(config.Index))
 	if err != nil {
 		return err
@@ -90,7 +90,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, service g
 	//case <-ctx.Done():
 	//}
 	<-apiCtx.Done()
-	exitCause := context.Cause(ctx)
+	exitCause := context.Cause(apiCtx)
 	log.ZWarn(ctx, "api server exit", exitCause)
 	timer := time.NewTimer(time.Second * 15)
 	defer timer.Stop()
@@ -2,9 +2,6 @@ package jssdk

 import (
 	"context"
-	"io"
-	"strings"
-
 	"github.com/gin-gonic/gin"
 	"github.com/openimsdk/tools/a2r"
 	"github.com/openimsdk/tools/apiresp"
@@ -12,6 +9,8 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
+	"io"
+	"strings"
 )

 func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A, opts ...grpc.CallOption) (*B, error), req *A, get func(*B) C) (C, error) {
@@ -94,7 +94,22 @@ func (*MessageApi) SetOptions(options map[string]bool, value bool) {
 	datautil.SetSwitchFromOptions(options, constant.IsConversationUpdate, value)
 }

-func (m *MessageApi) newUserSendMsgReq(_ *gin.Context, params *apistruct.SendMsg) *msg.SendMsgReq {
+func (m *MessageApi) newUserSendMsgReq(_ *gin.Context, params *apistruct.SendMsg, data any) *msg.SendMsgReq {
+	msgData := &sdkws.MsgData{
+		SendID:           params.SendID,
+		GroupID:          params.GroupID,
+		ClientMsgID:      idutil.GetMsgIDByMD5(params.SendID),
+		SenderPlatformID: params.SenderPlatformID,
+		SenderNickname:   params.SenderNickname,
+		SenderFaceURL:    params.SenderFaceURL,
+		SessionType:      params.SessionType,
+		MsgFrom:          constant.SysMsgType,
+		ContentType:      params.ContentType,
+		CreateTime:       timeutil.GetCurrentTimestampByMill(),
+		SendTime:         params.SendTime,
+		OfflinePushInfo:  params.OfflinePushInfo,
+		Ex:               params.Ex,
+	}
 	var newContent string
 	options := make(map[string]bool, 5)
 	switch params.ContentType {
@@ -104,6 +119,11 @@ func (m *MessageApi) newUserSendMsgReq(_ *gin.Context, params *apistruct.SendMsg
 		newContent = jsonutil.StructToJsonString(&notification)
 	case constant.Text:
 		fallthrough
+	case constant.AtText:
+		if atElem, ok := data.(*apistruct.AtElem); ok {
+			msgData.AtUserIDList = atElem.AtUserList
+		}
+		fallthrough
 	case constant.Picture:
 		fallthrough
 	case constant.Custom:
@@ -123,24 +143,10 @@ func (m *MessageApi) newUserSendMsgReq(_ *gin.Context, params *apistruct.SendMsg
 	if params.NotOfflinePush {
 		datautil.SetSwitchFromOptions(options, constant.IsOfflinePush, false)
 	}
+	msgData.Content = []byte(newContent)
+	msgData.Options = options
 	pbData := msg.SendMsgReq{
-		MsgData: &sdkws.MsgData{
-			SendID:           params.SendID,
-			GroupID:          params.GroupID,
-			ClientMsgID:      idutil.GetMsgIDByMD5(params.SendID),
-			SenderPlatformID: params.SenderPlatformID,
-			SenderNickname:   params.SenderNickname,
-			SenderFaceURL:    params.SenderFaceURL,
-			SessionType:      params.SessionType,
-			MsgFrom:          constant.SysMsgType,
-			ContentType:      params.ContentType,
-			Content:          []byte(newContent),
-			CreateTime:       timeutil.GetCurrentTimestampByMill(),
-			SendTime:         params.SendTime,
-			Options:          options,
-			OfflinePushInfo:  params.OfflinePushInfo,
-			Ex:               params.Ex,
-		},
+		MsgData: msgData,
 	}
 	return &pbData
 }
@@ -198,23 +204,23 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
 	log.ZDebug(c, "getSendMsgReq", "req", req.Content)
 	switch req.ContentType {
 	case constant.Text:
-		data = apistruct.TextElem{}
+		data = &apistruct.TextElem{}
 	case constant.Picture:
-		data = apistruct.PictureElem{}
+		data = &apistruct.PictureElem{}
 	case constant.Voice:
-		data = apistruct.SoundElem{}
+		data = &apistruct.SoundElem{}
 	case constant.Video:
-		data = apistruct.VideoElem{}
+		data = &apistruct.VideoElem{}
 	case constant.File:
-		data = apistruct.FileElem{}
+		data = &apistruct.FileElem{}
 	case constant.AtText:
-		data = apistruct.AtElem{}
+		data = &apistruct.AtElem{}
 	case constant.Custom:
-		data = apistruct.CustomElem{}
+		data = &apistruct.CustomElem{}
 	case constant.MarkdownText:
-		data = apistruct.MarkdownTextElem{}
+		data = &apistruct.MarkdownTextElem{}
 	case constant.OANotification:
-		data = apistruct.OANotificationElem{}
+		data = &apistruct.OANotificationElem{}
 		req.SessionType = constant.NotificationChatType
 		if err = m.userClient.GetNotificationByID(c, req.SendID); err != nil {
 			return nil, err
@@ -222,14 +228,14 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
 	default:
 		return nil, errs.WrapMsg(errs.ErrArgs, "unsupported content type", "contentType", req.ContentType)
 	}
-	if err := mapstructure.WeakDecode(req.Content, &data); err != nil {
+	if err := mapstructure.WeakDecode(req.Content, data); err != nil {
 		return nil, errs.WrapMsg(err, "failed to decode message content")
 	}
 	log.ZDebug(c, "getSendMsgReq", "decodedContent", data)
 	if err := m.validate.Struct(data); err != nil {
 		return nil, errs.WrapMsg(err, "validation error")
 	}
-	return m.newUserSendMsgReq(c, &req), nil
+	return m.newUserSendMsgReq(c, &req, data), nil
 }

 func (m *MessageApi) getModifyFields(req, respModify *sdkws.MsgData) map[string]any {
@@ -467,6 +473,10 @@ func (m *MessageApi) SendSimpleMessage(c *gin.Context) {
 		sessionType int32
 		recvID      string
 	)
+	if err = c.BindJSON(&req); err != nil {
+		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+		return
+	}
 	err = json.Unmarshal(decodedData, &keyMsgData)
 	if err != nil {
 		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
@@ -490,6 +500,11 @@ func (m *MessageApi) SendSimpleMessage(c *gin.Context) {
 		return
 	}

+	content, err := jsonutil.JsonMarshal(apistruct.MarkdownTextElem{Content: req.Content})
+	if err != nil {
+		apiresp.GinError(c, errs.Wrap(err))
+		return
+	}
 	msgData := &sdkws.MsgData{
 		SendID:           sendID,
 		RecvID:           recvID,
@@ -498,17 +513,17 @@ func (m *MessageApi) SendSimpleMessage(c *gin.Context) {
 		SenderPlatformID: constant.AdminPlatformID,
 		SessionType:      sessionType,
 		MsgFrom:          constant.UserMsgType,
-		ContentType:      constant.Text,
-		Content:          []byte(req.Content),
+		ContentType:      constant.MarkdownText,
+		Content:          content,
 		OfflinePushInfo:  req.OfflinePushInfo,
 		Ex:               req.Ex,
 	}

-	sendReq := &msg.SendMsgReq{
+	sendReq := &msg.SendSimpleMsgReq{
 		MsgData: msgData,
 	}

-	respPb, err := m.Client.SendMsg(c, sendReq)
+	respPb, err := m.Client.SendSimpleMsg(c, sendReq)
 	if err != nil {
 		apiresp.GinError(c, err)
 		return
@@ -525,7 +540,12 @@ func (m *MessageApi) SendSimpleMessage(c *gin.Context) {
 		return
 	}

-	m.ginRespSendMsg(c, sendReq, respPb)
+	m.ginRespSendMsg(c, &msg.SendMsgReq{MsgData: sendReq.MsgData}, &msg.SendMsgResp{
+		ServerMsgID: respPb.ServerMsgID,
+		ClientMsgID: respPb.ClientMsgID,
+		SendTime:    respPb.SendTime,
+		Modify:      respPb.Modify,
+	})
 }

 func (m *MessageApi) CheckMsgIsSendSuccess(c *gin.Context) {
@@ -6,35 +6,29 @@ import (
 	"net/http"

 	"github.com/gin-gonic/gin"
-	conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/apiresp"
 	"github.com/openimsdk/tools/discovery"
-	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/errs"
-	clientv3 "go.etcd.io/etcd/client/v3"
 )

 type PrometheusDiscoveryApi struct {
 	config *Config
-	client *clientv3.Client
 	kv     discovery.KeyValue
 }

-func NewPrometheusDiscoveryApi(config *Config, client discovery.Conn) *PrometheusDiscoveryApi {
+func NewPrometheusDiscoveryApi(config *Config, client discovery.SvcDiscoveryRegistry) *PrometheusDiscoveryApi {
 	api := &PrometheusDiscoveryApi{
 		config: config,
-	}
-	if config.Discovery.Enable == conf.ETCD {
-		api.client = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
+		kv:     client,
 	}
 	return api
 }

 func (p *PrometheusDiscoveryApi) discovery(c *gin.Context, key string) {
-	value, err := p.kv.GetKey(c, prommetrics.BuildDiscoveryKey(key))
+	value, err := p.kv.GetKeyWithPrefix(c, prommetrics.BuildDiscoveryKeyPrefix(key))
 	if err != nil {
-		if errors.Is(err, discovery.ErrNotSupportedKeyValue) {
+		if errors.Is(err, discovery.ErrNotSupported) {
 			c.JSON(http.StatusOK, []struct{}{})
 			return
 		}
@@ -46,10 +40,17 @@ func (p *PrometheusDiscoveryApi) discovery(c *gin.Context, key string) {
 		return
 	}
 	var resp prommetrics.RespTarget
-	if err := json.Unmarshal(value, &resp); err != nil {
-		apiresp.GinError(c, errs.WrapMsg(err, "json unmarshal err"))
-		return
+	for i := range value {
+		var tmp prommetrics.Target
+		if err = json.Unmarshal(value[i], &tmp); err != nil {
+			apiresp.GinError(c, errs.WrapMsg(err, "json unmarshal err"))
+			return
+		}
+
+		resp.Targets = append(resp.Targets, tmp.Target)
+		resp.Labels = tmp.Labels // default label is fixed. See prommetrics.BuildDefaultTarget
 	}
+
 	c.JSON(http.StatusOK, []*prommetrics.RespTarget{&resp})
 }

@@ -9,8 +9,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin/binding"
 	"github.com/go-playground/validator/v10"
-	clientv3 "go.etcd.io/etcd/client/v3"
-
 	"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -30,6 +28,8 @@ 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"
 )

 const (
@@ -54,7 +54,7 @@ func prommetricsGin() gin.HandlerFunc {
 	}
 }

-func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin.Engine, error) {
+func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cfg *Config) (*gin.Engine, error) {
 	authConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Auth)
 	if err != nil {
 		return nil, err
@@ -97,8 +97,8 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
-	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
-		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)), setGinIsAdmin(cfg.Share.IMAdminUserID))
+	r.Use(api.GinLogger(), prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
+		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)), setGinIsAdmin(cfg.Share.IMAdminUser.UserIDs))

 	u := NewUserApi(user.NewUserClient(userConn), client, cfg.Discovery.RpcService)
 	{
@@ -232,7 +232,7 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin
 		objectGroup.GET("/*name", t.ObjectRedirect)
 	}
 	// Message
-	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), cfg.Share.IMAdminUserID)
+	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), cfg.Share.IMAdminUser.UserIDs)
 	{
 		msgGroup := r.Group("/msg")
 		msgGroup.POST("/newest_seq", m.GetSeq)
@@ -253,6 +253,7 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin
 		msgGroup.POST("/delete_msg_physical", m.DeleteMsgPhysical)

 		msgGroup.POST("/batch_send_msg", m.BatchSendMsg)
+		msgGroup.POST("/send_simple_msg", m.SendSimpleMessage)
 		msgGroup.POST("/check_msg_is_send_success", m.CheckMsgIsSendSuccess)
 		msgGroup.POST("/get_server_time", m.GetServerTime)
 	}
@@ -309,13 +310,12 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin
 	if cfg.Discovery.Enable == config.ETCD {
 		etcdClient = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 	}
-	cm := NewConfigManager(cfg.Share.IMAdminUserID, &cfg.AllConfig, etcdClient, string(cfg.ConfigPath))
+	cm := NewConfigManager(cfg.Share.IMAdminUser.UserIDs, &cfg.AllConfig, etcdClient, string(cfg.ConfigPath))
 	{
 		configGroup := r.Group("/config", cm.CheckAdmin)
 		configGroup.POST("/get_config_list", cm.GetConfigList)
 		configGroup.POST("/get_config", cm.GetConfig)
 		configGroup.POST("/set_config", cm.SetConfig)
-		configGroup.POST("/set_configs", cm.SetConfigs)
 		configGroup.POST("/reset_config", cm.ResetConfig)
 		configGroup.POST("/set_enable_config_manager", cm.SetEnableConfigManager)
 		configGroup.POST("/get_enable_config_manager", cm.GetEnableConfigManager)
@@ -249,6 +249,7 @@ func (s *Server) MultiTerminalLoginCheck(ctx context.Context, req *msggateway.Mu
 		tempUserCtx.SetOperationID(mcontext.GetOperationID(ctx))
 		client := &Client{}
 		client.ctx = tempUserCtx
+		client.token = req.Token
 		client.UserID = req.UserID
 		client.PlatformID = int(req.PlatformID)
 		i := &kickHandler{
@@ -39,7 +39,7 @@ type Config struct {
 }

 // Start run ws server.
-func Start(ctx context.Context, conf *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	log.CInfo(ctx, "MSG-GATEWAY server is initializing", "runtimeEnv", runtimeenv.RuntimeEnvironment(),
 		"rpcPorts", conf.MsgGateway.RPC.Ports,
 		"wsPort", conf.MsgGateway.LongConnSvr.Ports, "prometheusPorts", conf.MsgGateway.Prometheus.Ports)
@@ -5,16 +5,17 @@ import (
 	"crypto/md5"
 	"encoding/binary"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
-	pbuser "github.com/openimsdk/protocol/user"
-	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mcontext"
-	"github.com/openimsdk/tools/utils/datautil"
 	"math/rand"
 	"os"
 	"strconv"
 	"sync/atomic"
 	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
+	pbuser "github.com/openimsdk/protocol/user"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mcontext"
+	"github.com/openimsdk/tools/utils/datautil"
 )

 func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
@@ -130,7 +130,7 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer {
 	for _, o := range opts {
 		o(&config)
 	}
-	//userRpcClient := rpcclient.NewUserRpcClient(client, config.Discovery.RpcService.User, config.Share.IMAdminUserID)
+	//userRpcClient := rpcclient.NewUserRpcClient(client, config.Discovery.RpcService.User, config.Share.IMAdminUser)

 	v := validator.New()
 	return &WsServer{
@@ -334,17 +334,51 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		}
 	}

+	// If reconnect: When multiple msgGateway instances are deployed, a client may disconnect from instance A and reconnect to instance B.
+	// During this process, instance A might still be executing, resulting in two clients with the same token existing simultaneously.
+	// This situation needs to be filtered to prevent duplicate clients.
+	checkSameTokenFunc := func(oldClients []*Client) []*Client {
+		var clientsNeedToKick []*Client
+
+		for _, c := range oldClients {
+			if c.token == newClient.token {
+				log.ZDebug(newClient.ctx, "token is same, not kick",
+					"userID", newClient.UserID,
+					"platformID", newClient.PlatformID,
+					"token", newClient.token)
+				continue
+			}
+
+			clientsNeedToKick = append(clientsNeedToKick, c)
+		}
+
+		return clientsNeedToKick
+	}
+
 	switch ws.msgGatewayConfig.Share.MultiLogin.Policy {
 	case constant.DefalutNotKick:
 	case constant.PCAndOther:
 		if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC {
 			return
 		}
+		clients, ok := ws.clients.GetAll(newClient.UserID)
+		clientOK = ok
+		oldClients = make([]*Client, 0, len(clients))
+		for _, c := range clients {
+			if constant.PlatformIDToClass(c.PlatformID) == constant.TerminalPC {
+				continue
+			}
+			oldClients = append(oldClients, c)
+		}
+
 		fallthrough
 	case constant.AllLoginButSameTermKick:
 		if !clientOK {
 			return
 		}
+
+		oldClients = checkSameTokenFunc(oldClients)
+
 		ws.clients.DeleteClients(newClient.UserID, oldClients)
 		for _, c := range oldClients {
 			err := c.KickOnlineMessage()
@@ -352,6 +386,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 				log.ZWarn(c.ctx, "KickOnlineMessage", err)
 			}
 		}
+
 		ctx := mcontext.WithMustInfoCtx(
 			[]string{newClient.ctx.GetOperationID(), newClient.ctx.GetUserID(),
 				constant.PlatformIDToName(newClient.PlatformID), newClient.ctx.GetConnID()},
@@ -370,14 +405,17 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		if !ok {
 			return
 		}
-		var (
-			kickClients []*Client
-		)
+
+		var kickClients []*Client
 		for _, client := range clients {
 			if constant.PlatformIDToClass(client.PlatformID) == constant.PlatformIDToClass(newClient.PlatformID) {
-				kickClients = append(kickClients, client)
+				{
+					kickClients = append(kickClients, client)
+				}
 			}
 		}
+		kickClients = checkSameTokenFunc(kickClients)
+
 		kickTokenFunc(kickClients)
 	}
 }
@@ -0,0 +1,132 @@
+package msgtransfer
+
+import (
+	"context"
+	"encoding/base64"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
+	"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"
+	"github.com/openimsdk/tools/utils/datautil"
+	"github.com/openimsdk/tools/utils/stringutil"
+	"google.golang.org/protobuf/proto"
+
+	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) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
+	if msg.ContentType == constant.Typing {
+		return
+	}
+
+	if !filterAfterMsg(msg, after) {
+		return
+	}
+
+	cbReq := &cbapi.CallbackAfterSendSingleMsgReq{
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand),
+		RecvID:            msg.RecvID,
+	}
+	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg))
+}
+
+func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) {
+	if msg.ContentType == constant.Typing {
+		return
+	}
+
+	if !filterAfterMsg(msg, after) {
+		return
+	}
+
+	cbReq := &cbapi.CallbackAfterSendGroupMsgReq{
+		CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand),
+		GroupID:           msg.GroupID,
+	}
+
+	mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg))
+}
+
+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)),
+	}
+}
+
+func filterAfterMsg(msg *sdkws.MsgData, after *config.AfterConfig) bool {
+	return filterMsg(msg, after.AttentionIds, after.DeniedTypes)
+}
+
+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...) {
+		return false
+	}
+
+	if defaultDeniedTypes(msg.ContentType) {
+		return false
+	}
+
+	if len(deniedTypes) != 0 && datautil.Contain(msg.ContentType, deniedTypes...) {
+		return false
+	}
+
+	return true
+}
+
+func defaultDeniedTypes(contentType int32) bool {
+	if contentType >= constant.NotificationBegin && contentType <= constant.NotificationEnd {
+		return true
+	}
+	if contentType == constant.Typing {
+		return true
+	}
+	return false
+}
@@ -58,7 +58,7 @@ type Config struct {
 	Index          conf.Index
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	builder := mqbuild.NewBuilder(&config.KafkaConfig)

 	log.CInfo(ctx, "MSG-TRANSFER server is initializing", "runTimeEnv", runtimeenv.RuntimeEnvironment(), "prometheusPorts",
@@ -134,7 +134,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	if err != nil {
 		return err
 	}
-	historyMongoHandler := NewOnlineHistoryMongoConsumerHandler(msgTransferDatabase)
+	historyMongoHandler := NewOnlineHistoryMongoConsumerHandler(msgTransferDatabase, config)

 	msgTransfer := &MsgTransfer{
 		historyConsumer:      historyConsumer,
@@ -161,8 +161,8 @@ func (m *MsgTransfer) Start(ctx context.Context) error {
 	}()

 	go func() {
-		fn := func(ctx context.Context, key string, value []byte) error {
-			m.historyMongoHandler.HandleChatWs2Mongo(ctx, key, value)
+		fn := func(msg mq.Message) error {
+			m.historyMongoHandler.HandleChatWs2Mongo(msg)
 			return nil
 		}
 		for {
@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/openimsdk/tools/mq"

 	"sync"
 	"time"
@@ -26,6 +27,8 @@ import (
 	"github.com/openimsdk/tools/discovery"

 	"github.com/go-redis/redis"
+	"google.golang.org/protobuf/proto"
+
 	"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/msgprocessor"
@@ -37,7 +40,6 @@ import (
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/stringutil"
-	"google.golang.org/protobuf/proto"
 )

 const (
@@ -76,6 +78,7 @@ type ConsumerMessage struct {
 	Ctx   context.Context
 	Key   string
 	Value []byte
+	Raw   mq.Message
 }

 func NewOnlineHistoryRedisConsumerHandler(ctx context.Context, client discovery.Conn, config *Config, database controller.MsgTransferDatabase) (*OnlineHistoryRedisConsumerHandler, error) {
@@ -112,6 +115,11 @@ func NewOnlineHistoryRedisConsumerHandler(ctx context.Context, client discovery.
 	b.Do = och.do
 	och.redisMessageBatches = b

+	och.redisMessageBatches.OnComplete = func(lastMessage *ConsumerMessage, totalCount int) {
+		lastMessage.Raw.Mark()
+		lastMessage.Raw.Commit()
+	}
+
 	return &och, nil
 }
 func (och *OnlineHistoryRedisConsumerHandler) do(ctx context.Context, channelID int, val *batcher.Msg[ConsumerMessage]) {
@@ -134,53 +142,48 @@ func (och *OnlineHistoryRedisConsumerHandler) do(ctx context.Context, channelID

 func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context, msgs []*ContextMsg) {

-	var conversationID string
-	var userSeqMap map[string]int64
+	// Outer map: conversationID -> (userID -> maxHasReadSeq)
+	conversationUserSeq := make(map[string]map[string]int64)
+
 	for _, msg := range msgs {
 		if msg.message.ContentType != constant.HasReadReceipt {
 			continue
 		}
 		var elem sdkws.NotificationElem
 		if err := json.Unmarshal(msg.message.Content, &elem); err != nil {
-			log.ZWarn(ctx, "handlerConversationRead Unmarshal NotificationElem msg err", err, "msg", msg)
+			log.ZWarn(ctx, "Unmarshal NotificationElem error", err, "msg", msg)
 			continue
 		}
 		var tips sdkws.MarkAsReadTips
 		if err := json.Unmarshal([]byte(elem.Detail), &tips); err != nil {
-			log.ZWarn(ctx, "handlerConversationRead Unmarshal MarkAsReadTips msg err", err, "msg", msg)
+			log.ZWarn(ctx, "Unmarshal MarkAsReadTips error", err, "msg", msg)
 			continue
 		}
-		//The conversation ID for each batch of messages processed by the batcher is the same.
-		conversationID = tips.ConversationID
-		if len(tips.Seqs) > 0 {
-			for _, seq := range tips.Seqs {
-				if tips.HasReadSeq < seq {
-					tips.HasReadSeq = seq
-				}
-			}
-			clear(tips.Seqs)
-			tips.Seqs = nil
-		}
-		if tips.HasReadSeq < 0 {
+		if len(tips.ConversationID) == 0 || tips.HasReadSeq < 0 {
 			continue
 		}
-		if userSeqMap == nil {
-			userSeqMap = make(map[string]int64)
-		}

-		if userSeqMap[tips.MarkAsReadUserID] > tips.HasReadSeq {
-			continue
+		// Calculate the max seq from tips.Seqs
+		for _, seq := range tips.Seqs {
+			if tips.HasReadSeq < seq {
+				tips.HasReadSeq = seq
+			}
+		}
+
+		if _, ok := conversationUserSeq[tips.ConversationID]; !ok {
+			conversationUserSeq[tips.ConversationID] = make(map[string]int64)
+		}
+		if conversationUserSeq[tips.ConversationID][tips.MarkAsReadUserID] < tips.HasReadSeq {
+			conversationUserSeq[tips.ConversationID][tips.MarkAsReadUserID] = tips.HasReadSeq
 		}
-		userSeqMap[tips.MarkAsReadUserID] = tips.HasReadSeq
 	}
-	if userSeqMap == nil {
-		return
-	}
-	if len(conversationID) == 0 {
-		log.ZWarn(ctx, "conversation err", nil, "conversationID", conversationID)
-	}
-	if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, conversationID, userSeqMap); err != nil {
-		log.ZWarn(ctx, "set read seq to db error", err, "conversationID", conversationID, "userSeqMap", userSeqMap)
+	log.ZInfo(ctx, "doSetReadSeq", "conversationUserSeq", conversationUserSeq)
+
+	// persist to db
+	for convID, userSeqMap := range conversationUserSeq {
+		if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, convID, userSeqMap); err != nil {
+			log.ZWarn(ctx, "SetHasReadSeqToDB error", err, "conversationID", convID, "userSeqMap", userSeqMap)
+		}
 	}

 }
@@ -392,10 +395,10 @@ func withAggregationCtx(ctx context.Context, values []*ContextMsg) context.Conte
 	return mcontext.SetOperationID(ctx, allMessageOperationID)
 }

-func (och *OnlineHistoryRedisConsumerHandler) HandlerRedisMessage(ctx context.Context, key string, value []byte) error { // a instance in the consumer group
-	err := och.redisMessageBatches.Put(ctx, &ConsumerMessage{Ctx: ctx, Key: key, Value: value})
+func (och *OnlineHistoryRedisConsumerHandler) HandlerRedisMessage(msg mq.Message) error { // a instance in the consumer group
+	err := och.redisMessageBatches.Put(msg.Context(), &ConsumerMessage{Ctx: msg.Context(), Key: msg.Key(), Value: msg.Value(), Raw: msg})
 	if err != nil {
-		log.ZWarn(ctx, "put msg to  error", err, "key", key, "value", value)
+		log.ZWarn(msg.Context(), "put msg to  error", err, "key", msg.Key(), "value", msg.Value())
 	}
 	return nil
 }
@@ -15,10 +15,12 @@
 package msgtransfer

 import (
-	"context"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/tools/mq"

 	"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/webhook"
 	pbmsg "github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/tools/log"
 	"google.golang.org/protobuf/proto"
@@ -26,15 +28,22 @@ import (

 type OnlineHistoryMongoConsumerHandler struct {
 	msgTransferDatabase controller.MsgTransferDatabase
+	config              *Config
+	webhookClient       *webhook.Client
 }

-func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase) *OnlineHistoryMongoConsumerHandler {
+func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabase, config *Config) *OnlineHistoryMongoConsumerHandler {
 	return &OnlineHistoryMongoConsumerHandler{
 		msgTransferDatabase: database,
+		config:              config,
+		webhookClient:       webhook.NewWebhookClient(config.WebhooksConfig.URL),
 	}
 }

-func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Context, key string, msg []byte) {
+func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) {
+	ctx := val.Context()
+	key := val.Key()
+	msg := val.Value()
 	msgFromMQ := pbmsg.MsgDataToMongoByMQ{}
 	err := proto.Unmarshal(msg, &msgFromMQ)
 	if err != nil {
@@ -52,7 +61,18 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont
 		prommetrics.MsgInsertMongoFailedCounter.Inc()
 	} else {
 		prommetrics.MsgInsertMongoSuccessCounter.Inc()
+		val.Mark()
 	}
+
+	for _, msgData := range msgFromMQ.MsgData {
+		switch msgData.SessionType {
+		case constant.SingleChatType:
+			mc.webhookAfterSendSingleMsg(ctx, &mc.config.WebhooksConfig.AfterSendSingleMsg, msgData)
+		case constant.ReadGroupChatType:
+			mc.webhookAfterSendGroupMsg(ctx, &mc.config.WebhooksConfig.AfterSendGroupMsg, msgData)
+		}
+	}
+
 	//var seqs []int64
 	//for _, msg := range msgFromMQ.MsgData {
 	//	seqs = append(seqs, msg.Seq)
@@ -2,6 +2,7 @@ package push

 import (
 	"context"
+	"github.com/openimsdk/tools/mq"
 	"math/rand"
 	"strconv"

@@ -50,7 +51,7 @@ func (p pushServer) DelUserPushToken(ctx context.Context,
 	return &pbpush.DelUserPushTokenResp{}, nil
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongoConfig, &config.RedisConfig)
 	rdb, err := dbb.Redis(ctx)
 	if err != nil {
@@ -106,8 +107,8 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr

 	go func() {
 		pushHandler.WaitCache()
-		fn := func(ctx context.Context, key string, value []byte) error {
-			pushHandler.HandleMs2PsChat(authverify.WithTempAdmin(ctx), value)
+		fn := func(msg mq.Message) error {
+			pushHandler.HandleMs2PsChat(authverify.WithTempAdmin(msg.Context()), msg.Value())
 			return nil
 		}
 		consumerCtx := mcontext.SetOperationID(context.Background(), "push_"+strconv.Itoa(int(rand.Uint32())))
@@ -121,8 +122,8 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	}()

 	go func() {
-		fn := func(ctx context.Context, key string, value []byte) error {
-			offlineHandler.HandleMsg2OfflinePush(ctx, value)
+		fn := func(msg mq.Message) error {
+			offlineHandler.HandleMsg2OfflinePush(msg.Context(), msg.Value())
 			return nil
 		}
 		consumerCtx := mcontext.SetOperationID(context.Background(), "push_"+strconv.Itoa(int(rand.Uint32())))
@@ -317,8 +317,8 @@ func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID stri
 					return err
 				}
 				log.ZDebug(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(*pushToUserIDs), "list", pushToUserIDs)
-				if len(c.config.Share.IMAdminUserID) > 0 {
-					ctx = mcontext.WithOpUserIDContext(ctx, c.config.Share.IMAdminUserID[0])
+				if len(c.config.Share.IMAdminUser.UserIDs) > 0 {
+					ctx = mcontext.WithOpUserIDContext(ctx, c.config.Share.IMAdminUser.UserIDs[0])
 				}
 				defer func(groupID string) {
 					if err := c.groupClient.DismissGroup(ctx, groupID, true); err != nil {
@@ -49,6 +49,7 @@ type authServer struct {
 	RegisterCenter discovery.Conn
 	config         *Config
 	userClient     *rpcli.UserClient
+	adminUserIDs   []string
 }

 type Config struct {
@@ -59,7 +60,7 @@ type Config struct {
 	Discovery   config.Discovery
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongoConfig, &config.RedisConfig)
 	rdb, err := dbb.Redis(ctx)
 	if err != nil {
@@ -90,10 +91,11 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 			config.Share.Secret,
 			config.RpcConfig.TokenPolicy.Expire,
 			config.Share.MultiLogin,
-			config.Share.IMAdminUserID,
+			config.Share.IMAdminUser.UserIDs,
 		),
-		config:     config,
-		userClient: rpcli.NewUserClient(userConn),
+		config:       config,
+		userClient:   rpcli.NewUserClient(userConn),
+		adminUserIDs: config.Share.IMAdminUser.UserIDs,
 	})
 	return nil
 }
@@ -104,8 +106,8 @@ func (s *authServer) GetAdminToken(ctx context.Context, req *pbauth.GetAdminToke
 		return nil, errs.ErrNoPermission.WrapMsg("secret invalid")
 	}

-	if !datautil.Contain(req.UserID, s.config.Share.IMAdminUserID...) {
-		return nil, errs.ErrArgs.WrapMsg("userID is error.", "userID", req.UserID, "adminUserID", s.config.Share.IMAdminUserID)
+	if !datautil.Contain(req.UserID, s.adminUserIDs...) {
+		return nil, errs.ErrArgs.WrapMsg("userID is error.", "userID", req.UserID, "adminUserID", s.adminUserIDs)

 	}

@@ -31,7 +31,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	dbModel "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
@@ -69,7 +68,7 @@ type Config struct {
 	Discovery          config.Discovery
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig)
 	mgocli, err := dbb.Mongo(ctx)
 	if err != nil {
@@ -244,11 +243,14 @@ func (c *conversationServer) getConversations(ctx context.Context, ownerUserID s
 	return convert.ConversationsDB2Pb(conversations), nil
 }

+// Deprecated
 func (c *conversationServer) SetConversation(ctx context.Context, req *pbconversation.SetConversationReq) (*pbconversation.SetConversationResp, error) {
 	if err := authverify.CheckAccess(ctx, req.GetConversation().GetUserID()); err != nil {
 		return nil, err
 	}
 	var conversation dbModel.Conversation
+	conversation.CreateTime = time.Now()
+
 	if err := datautil.CopyStructFields(&conversation, req.Conversation); err != nil {
 		return nil, err
 	}
@@ -300,6 +302,7 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
 	conversation.ConversationType = req.Conversation.ConversationType
 	conversation.UserID = req.Conversation.UserID
 	conversation.GroupID = req.Conversation.GroupID
+	conversation.CreateTime = time.Now()

 	m, conversation, err := UpdateConversationsMap(ctx, req)
 	if err != nil {
@@ -365,6 +368,8 @@ func (c *conversationServer) UpdateConversationsByUser(ctx context.Context, req
 // create conversation without notification for msg redis transfer.
 func (c *conversationServer) CreateSingleChatConversations(ctx context.Context, req *pbconversation.CreateSingleChatConversationsReq) (*pbconversation.CreateSingleChatConversationsResp, error) {
 	var conversation dbModel.Conversation
+	conversation.CreateTime = time.Now()
+
 	switch req.ConversationType {
 	case constant.SingleChatType:
 		// sendUser create
@@ -372,6 +377,7 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 		conversation.ConversationType = req.ConversationType
 		conversation.OwnerUserID = req.SendID
 		conversation.UserID = req.RecvID
+
 		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
 			return nil, err
 		}
@@ -387,6 +393,7 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 		conversation2 := conversation
 		conversation2.OwnerUserID = req.RecvID
 		conversation2.UserID = req.SendID
+
 		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
 			return nil, err
 		}
@@ -402,6 +409,7 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 		conversation.ConversationType = req.ConversationType
 		conversation.OwnerUserID = req.RecvID
 		conversation.UserID = req.SendID
+
 		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
 			return nil, err
 		}
@@ -423,6 +431,7 @@ func (c *conversationServer) CreateGroupChatConversations(ctx context.Context, r
 	conversation.ConversationID = msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
 	conversation.GroupID = req.GroupID
 	conversation.ConversationType = constant.ReadGroupChatType
+	conversation.CreateTime = time.Now()

 	if err := c.webhookBeforeCreateGroupChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateGroupChatConversations, &conversation); err != nil {
 		return nil, err
@@ -708,7 +717,7 @@ func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _

 	maxPage := (num + batchNum - 1) / batchNum

-	temp := make([]*model.Conversation, 0, maxPage*batchNum)
+	temp := make([]*dbModel.Conversation, 0, maxPage*batchNum)

 	for pageNumber := 0; pageNumber < int(maxPage); pageNumber++ {
 		pagination := &sdkws.RequestPagination{
@@ -32,7 +32,7 @@ import (
 )

 // CallbackBeforeCreateGroup callback before create group.
-func (s *groupServer) webhookBeforeCreateGroup(ctx context.Context, before *config.BeforeConfig, req *group.CreateGroupReq) error {
+func (g *groupServer) webhookBeforeCreateGroup(ctx context.Context, before *config.BeforeConfig, req *group.CreateGroupReq) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		cbReq := &callbackstruct.CallbackBeforeCreateGroupReq{
 			CallbackCommand: callbackstruct.CallbackBeforeCreateGroupCommand,
@@ -57,7 +57,7 @@ func (s *groupServer) webhookBeforeCreateGroup(ctx context.Context, before *conf
 		}
 		resp := &callbackstruct.CallbackBeforeCreateGroupResp{}

-		if err := s.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
 			return err
 		}

@@ -77,7 +77,7 @@ func (s *groupServer) webhookBeforeCreateGroup(ctx context.Context, before *conf
 	})
 }

-func (s *groupServer) webhookAfterCreateGroup(ctx context.Context, after *config.AfterConfig, req *group.CreateGroupReq) {
+func (g *groupServer) webhookAfterCreateGroup(ctx context.Context, after *config.AfterConfig, req *group.CreateGroupReq) {
 	cbReq := &callbackstruct.CallbackAfterCreateGroupReq{
 		CallbackCommand: callbackstruct.CallbackAfterCreateGroupCommand,
 		GroupInfo:       req.GroupInfo,
@@ -98,10 +98,10 @@ func (s *groupServer) webhookAfterCreateGroup(ctx context.Context, after *config
 			RoleLevel: constant.GroupOrdinaryUsers,
 		})
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterCreateGroupResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterCreateGroupResp{}, after)
 }

-func (s *groupServer) webhookBeforeMembersJoinGroup(ctx context.Context, before *config.BeforeConfig, groupMembers []*model.GroupMember, groupID string, groupEx string) error {
+func (g *groupServer) webhookBeforeMembersJoinGroup(ctx context.Context, before *config.BeforeConfig, groupMembers []*model.GroupMember, groupID string, groupEx string) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		groupMembersMap := datautil.SliceToMap(groupMembers, func(e *model.GroupMember) string {
 			return e.UserID
@@ -123,7 +123,7 @@ func (s *groupServer) webhookBeforeMembersJoinGroup(ctx context.Context, before
 		}
 		resp := &callbackstruct.CallbackBeforeMembersJoinGroupResp{}

-		if err := s.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
 			return err
 		}

@@ -144,7 +144,7 @@ func (s *groupServer) webhookBeforeMembersJoinGroup(ctx context.Context, before
 	})
 }

-func (s *groupServer) webhookBeforeSetGroupMemberInfo(ctx context.Context, before *config.BeforeConfig, req *group.SetGroupMemberInfo) error {
+func (g *groupServer) webhookBeforeSetGroupMemberInfo(ctx context.Context, before *config.BeforeConfig, req *group.SetGroupMemberInfo) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		cbReq := callbackstruct.CallbackBeforeSetGroupMemberInfoReq{
 			CallbackCommand: callbackstruct.CallbackBeforeSetGroupMemberInfoCommand,
@@ -164,7 +164,7 @@ func (s *groupServer) webhookBeforeSetGroupMemberInfo(ctx context.Context, befor
 			cbReq.Ex = &req.Ex.Value
 		}
 		resp := &callbackstruct.CallbackBeforeSetGroupMemberInfoResp{}
-		if err := s.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
 			return err
 		}
 		if resp.FaceURL != nil {
@@ -183,7 +183,7 @@ func (s *groupServer) webhookBeforeSetGroupMemberInfo(ctx context.Context, befor
 	})
 }

-func (s *groupServer) webhookAfterSetGroupMemberInfo(ctx context.Context, after *config.AfterConfig, req *group.SetGroupMemberInfo) {
+func (g *groupServer) webhookAfterSetGroupMemberInfo(ctx context.Context, after *config.AfterConfig, req *group.SetGroupMemberInfo) {
 	cbReq := callbackstruct.CallbackAfterSetGroupMemberInfoReq{
 		CallbackCommand: callbackstruct.CallbackAfterSetGroupMemberInfoCommand,
 		GroupID:         req.GroupID,
@@ -201,55 +201,55 @@ func (s *groupServer) webhookAfterSetGroupMemberInfo(ctx context.Context, after
 	if req.Ex != nil {
 		cbReq.Ex = &req.Ex.Value
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterSetGroupMemberInfoResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterSetGroupMemberInfoResp{}, after)
 }

-func (s *groupServer) webhookAfterQuitGroup(ctx context.Context, after *config.AfterConfig, req *group.QuitGroupReq) {
+func (g *groupServer) webhookAfterQuitGroup(ctx context.Context, after *config.AfterConfig, req *group.QuitGroupReq) {
 	cbReq := &callbackstruct.CallbackQuitGroupReq{
 		CallbackCommand: callbackstruct.CallbackAfterQuitGroupCommand,
 		GroupID:         req.GroupID,
 		UserID:          req.UserID,
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackQuitGroupResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackQuitGroupResp{}, after)
 }

-func (s *groupServer) webhookAfterKickGroupMember(ctx context.Context, after *config.AfterConfig, req *group.KickGroupMemberReq) {
+func (g *groupServer) webhookAfterKickGroupMember(ctx context.Context, after *config.AfterConfig, req *group.KickGroupMemberReq) {
 	cbReq := &callbackstruct.CallbackKillGroupMemberReq{
 		CallbackCommand: callbackstruct.CallbackAfterKickGroupCommand,
 		GroupID:         req.GroupID,
 		KickedUserIDs:   req.KickedUserIDs,
 		Reason:          req.Reason,
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackKillGroupMemberResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackKillGroupMemberResp{}, after)
 }

-func (s *groupServer) webhookAfterDismissGroup(ctx context.Context, after *config.AfterConfig, req *callbackstruct.CallbackDisMissGroupReq) {
+func (g *groupServer) webhookAfterDismissGroup(ctx context.Context, after *config.AfterConfig, req *callbackstruct.CallbackDisMissGroupReq) {
 	req.CallbackCommand = callbackstruct.CallbackAfterDisMissGroupCommand
-	s.webhookClient.AsyncPost(ctx, req.GetCallbackCommand(), req, &callbackstruct.CallbackDisMissGroupResp{}, after)
+	g.webhookClient.AsyncPost(ctx, req.GetCallbackCommand(), req, &callbackstruct.CallbackDisMissGroupResp{}, after)
 }

-func (s *groupServer) webhookBeforeApplyJoinGroup(ctx context.Context, before *config.BeforeConfig, req *callbackstruct.CallbackJoinGroupReq) (err error) {
+func (g *groupServer) webhookBeforeApplyJoinGroup(ctx context.Context, before *config.BeforeConfig, req *callbackstruct.CallbackJoinGroupReq) (err error) {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		req.CallbackCommand = callbackstruct.CallbackBeforeJoinGroupCommand
 		resp := &callbackstruct.CallbackJoinGroupResp{}
-		if err := s.webhookClient.SyncPost(ctx, req.GetCallbackCommand(), req, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, req.GetCallbackCommand(), req, resp, before); err != nil {
 			return err
 		}
 		return nil
 	})
 }

-func (s *groupServer) webhookAfterTransferGroupOwner(ctx context.Context, after *config.AfterConfig, req *group.TransferGroupOwnerReq) {
+func (g *groupServer) webhookAfterTransferGroupOwner(ctx context.Context, after *config.AfterConfig, req *group.TransferGroupOwnerReq) {
 	cbReq := &callbackstruct.CallbackTransferGroupOwnerReq{
 		CallbackCommand: callbackstruct.CallbackAfterTransferGroupOwnerCommand,
 		GroupID:         req.GroupID,
 		OldOwnerUserID:  req.OldOwnerUserID,
 		NewOwnerUserID:  req.NewOwnerUserID,
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackTransferGroupOwnerResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackTransferGroupOwnerResp{}, after)
 }

-func (s *groupServer) webhookBeforeInviteUserToGroup(ctx context.Context, before *config.BeforeConfig, req *group.InviteUserToGroupReq) (err error) {
+func (g *groupServer) webhookBeforeInviteUserToGroup(ctx context.Context, before *config.BeforeConfig, req *group.InviteUserToGroupReq) (err error) {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		cbReq := &callbackstruct.CallbackBeforeInviteUserToGroupReq{
 			CallbackCommand: callbackstruct.CallbackBeforeInviteJoinGroupCommand,
@@ -260,7 +260,7 @@ func (s *groupServer) webhookBeforeInviteUserToGroup(ctx context.Context, before
 		}

 		resp := &callbackstruct.CallbackBeforeInviteUserToGroupResp{}
-		if err := s.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
 			return err
 		}

@@ -275,7 +275,7 @@ func (s *groupServer) webhookBeforeInviteUserToGroup(ctx context.Context, before
 	})
 }

-func (s *groupServer) webhookAfterJoinGroup(ctx context.Context, after *config.AfterConfig, req *group.JoinGroupReq) {
+func (g *groupServer) webhookAfterJoinGroup(ctx context.Context, after *config.AfterConfig, req *group.JoinGroupReq) {
 	cbReq := &callbackstruct.CallbackAfterJoinGroupReq{
 		CallbackCommand: callbackstruct.CallbackAfterJoinGroupCommand,
 		OperationID:     mcontext.GetOperationID(ctx),
@@ -284,10 +284,10 @@ func (s *groupServer) webhookAfterJoinGroup(ctx context.Context, after *config.A
 		JoinSource:      req.JoinSource,
 		InviterUserID:   req.InviterUserID,
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterJoinGroupResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterJoinGroupResp{}, after)
 }

-func (s *groupServer) webhookBeforeSetGroupInfo(ctx context.Context, before *config.BeforeConfig, req *group.SetGroupInfoReq) error {
+func (g *groupServer) webhookBeforeSetGroupInfo(ctx context.Context, before *config.BeforeConfig, req *group.SetGroupInfoReq) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		cbReq := &callbackstruct.CallbackBeforeSetGroupInfoReq{
 			CallbackCommand: callbackstruct.CallbackBeforeSetGroupInfoCommand,
@@ -312,7 +312,7 @@ func (s *groupServer) webhookBeforeSetGroupInfo(ctx context.Context, before *con
 		}
 		resp := &callbackstruct.CallbackBeforeSetGroupInfoResp{}

-		if err := s.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
 			return err
 		}

@@ -336,7 +336,7 @@ func (s *groupServer) webhookBeforeSetGroupInfo(ctx context.Context, before *con
 	})
 }

-func (s *groupServer) webhookAfterSetGroupInfo(ctx context.Context, after *config.AfterConfig, req *group.SetGroupInfoReq) {
+func (g *groupServer) webhookAfterSetGroupInfo(ctx context.Context, after *config.AfterConfig, req *group.SetGroupInfoReq) {
 	cbReq := &callbackstruct.CallbackAfterSetGroupInfoReq{
 		CallbackCommand: callbackstruct.CallbackAfterSetGroupInfoCommand,
 		GroupID:         req.GroupInfoForSet.GroupID,
@@ -357,10 +357,10 @@ func (s *groupServer) webhookAfterSetGroupInfo(ctx context.Context, after *confi
 	if req.GroupInfoForSet.ApplyMemberFriend != nil {
 		cbReq.ApplyMemberFriend = &req.GroupInfoForSet.ApplyMemberFriend.Value
 	}
-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterSetGroupInfoResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterSetGroupInfoResp{}, after)
 }

-func (s *groupServer) webhookBeforeSetGroupInfoEx(ctx context.Context, before *config.BeforeConfig, req *group.SetGroupInfoExReq) error {
+func (g *groupServer) webhookBeforeSetGroupInfoEx(ctx context.Context, before *config.BeforeConfig, req *group.SetGroupInfoExReq) error {
 	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
 		cbReq := &callbackstruct.CallbackBeforeSetGroupInfoExReq{
 			CallbackCommand: callbackstruct.CallbackBeforeSetGroupInfoExCommand,
@@ -388,7 +388,7 @@ func (s *groupServer) webhookBeforeSetGroupInfoEx(ctx context.Context, before *c

 		resp := &callbackstruct.CallbackBeforeSetGroupInfoExResp{}

-		if err := s.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+		if err := g.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
 			return err
 		}

@@ -405,7 +405,7 @@ func (s *groupServer) webhookBeforeSetGroupInfoEx(ctx context.Context, before *c
 	})
 }

-func (s *groupServer) webhookAfterSetGroupInfoEx(ctx context.Context, after *config.AfterConfig, req *group.SetGroupInfoExReq) {
+func (g *groupServer) webhookAfterSetGroupInfoEx(ctx context.Context, after *config.AfterConfig, req *group.SetGroupInfoExReq) {
 	cbReq := &callbackstruct.CallbackAfterSetGroupInfoExReq{
 		CallbackCommand: callbackstruct.CallbackAfterSetGroupInfoExCommand,
 		GroupID:         req.GroupID,
@@ -428,5 +428,5 @@ func (s *groupServer) webhookAfterSetGroupInfoEx(ctx context.Context, after *con
 		cbReq.ApplyMemberFriend = req.ApplyMemberFriend
 	}

-	s.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterSetGroupInfoExResp{}, after)
+	g.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterSetGroupInfoExResp{}, after)
 }
@@ -19,7 +19,7 @@ import (
 	"github.com/openimsdk/protocol/sdkws"
 )

-func (s *groupServer) groupDB2PB(group *model.Group, ownerUserID string, memberCount uint32) *sdkws.GroupInfo {
+func (g *groupServer) groupDB2PB(group *model.Group, ownerUserID string, memberCount uint32) *sdkws.GroupInfo {
 	return &sdkws.GroupInfo{
 		GroupID:                group.GroupID,
 		GroupName:              group.GroupName,
@@ -41,7 +41,7 @@ func (s *groupServer) groupDB2PB(group *model.Group, ownerUserID string, memberC
 	}
 }

-func (s *groupServer) groupMemberDB2PB(member *model.GroupMember, appMangerLevel int32) *sdkws.GroupMemberFullInfo {
+func (g *groupServer) groupMemberDB2PB(member *model.GroupMember, appMangerLevel int32) *sdkws.GroupMemberFullInfo {
 	return &sdkws.GroupMemberFullInfo{
 		GroupID:        member.GroupID,
 		UserID:         member.UserID,
@@ -58,6 +58,6 @@ func (s *groupServer) groupMemberDB2PB(member *model.GroupMember, appMangerLevel
 	}
 }

-func (s *groupServer) groupMemberDB2PB2(member *model.GroupMember) *sdkws.GroupMemberFullInfo {
-	return s.groupMemberDB2PB(member, 0)
+func (g *groupServer) groupMemberDB2PB2(member *model.GroupMember) *sdkws.GroupMemberFullInfo {
+	return g.groupMemberDB2PB(member, 0)
 }
@@ -16,9 +16,10 @@ package group

 import (
 	"context"
+
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 )

-func (s *groupServer) PopulateGroupMember(ctx context.Context, members ...*relationtb.GroupMember) error {
-	return s.notification.PopulateGroupMember(ctx, members...)
+func (g *groupServer) PopulateGroupMember(ctx context.Context, members ...*relationtb.GroupMember) error {
+	return g.notification.PopulateGroupMember(ctx, members...)
 }
@@ -63,6 +63,7 @@ type groupServer struct {
 	userClient         *rpcli.UserClient
 	msgClient          *rpcli.MsgClient
 	conversationClient *rpcli.ConversationClient
+	adminUserIDs       []string
 }

 type Config struct {
@@ -76,7 +77,7 @@ type Config struct {
 	Discovery          config.Discovery
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig)
 	mgocli, err := dbb.Mongo(ctx)
 	if err != nil {
@@ -116,6 +117,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 		userClient:         rpcli.NewUserClient(userConn),
 		msgClient:          rpcli.NewMsgClient(msgConn),
 		conversationClient: rpcli.NewConversationClient(conversationConn),
+		adminUserIDs:       config.Share.IMAdminUser.UserIDs,
 	}
 	gs.db = controller.NewGroupDatabase(rdb, &config.LocalCacheConfig, groupDB, groupMemberDB, groupRequestDB, mgocli.GetTx(), grouphash.NewGroupHashFromGroupServer(&gs))
 	gs.notification = NewNotificationSender(gs.db, config, gs.userClient, gs.msgClient, gs.conversationClient)
@@ -386,9 +388,9 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 	}

 	var groupMember *model.GroupMember
-	var opUserID string
+	opUserID := mcontext.GetOpUserID(ctx)
+
 	if !authverify.IsAdmin(ctx) {
-		opUserID = mcontext.GetOpUserID(ctx)
 		var err error
 		groupMember, err = g.db.TakeGroupMember(ctx, req.GroupID, opUserID)
 		if err != nil {
@@ -397,8 +399,6 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 		if err := g.PopulateGroupMember(ctx, groupMember); err != nil {
 			return nil, err
 		}
-	} else {
-		opUserID = mcontext.GetOpUserID(ctx)
 	}

 	if err := g.webhookBeforeInviteUserToGroup(ctx, &g.config.WebhooksConfig.BeforeInviteUserToGroup, req); err != nil && err != servererrs.ErrCallbackContinue {
@@ -457,10 +457,7 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite

 	const singleQuantity = 50
 	for start := 0; start < len(groupMembers); start += singleQuantity {
-		end := start + singleQuantity
-		if end > len(groupMembers) {
-			end = len(groupMembers)
-		}
+		end := min(start+singleQuantity, len(groupMembers))
 		currentMembers := groupMembers[start:end]

 		if err := g.db.CreateGroup(ctx, nil, currentMembers); err != nil {
@@ -471,8 +468,8 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 			return e.UserID
 		})

-		if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, userIDs...); err != nil {
-			return nil, err
+		if len(userIDs) != 0 {
+			g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, userIDs...)
 		}
 	}
 	return &pbgroup.InviteUserToGroupResp{}, nil
@@ -1906,7 +1903,7 @@ func (g *groupServer) GetSpecifiedUserGroupRequestInfo(ctx context.Context, req
 		}

 		adminIDs = append(adminIDs, owners[0].UserID)
-		adminIDs = append(adminIDs, g.config.Share.IMAdminUserID...)
+		adminIDs = append(adminIDs, g.adminUserIDs...)

 		if !datautil.Contain(opUserID, adminIDs...) {
 			return nil, errs.ErrNoPermission.WrapMsg("opUser no permission")
@@ -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)),
+// 	}
+// }
@@ -109,8 +109,8 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	revokerUserID := mcontext.GetOpUserID(ctx)
 	var flag bool

-	if len(m.config.Share.IMAdminUserID) > 0 {
-		flag = datautil.Contain(revokerUserID, m.config.Share.IMAdminUserID...)
+	if len(m.config.Share.IMAdminUser.UserIDs) > 0 {
+		flag = datautil.Contain(revokerUserID, m.adminUserIDs...)
 	}
 	tips := sdkws.RevokeMsgTips{
 		RevokerUserID:  revokerUserID,
@@ -86,7 +86,8 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq,
 		go m.setConversationAtInfo(ctx, req.MsgData)
 	}

-	m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req)
+	// m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req)
+
 	prommetrics.GroupChatMsgProcessSuccessCounter.Inc()
 	resp = &pbmsg.SendMsgResp{}
 	resp.SendTime = req.MsgData.SendTime
@@ -192,7 +193,8 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 			prommetrics.SingleChatMsgProcessFailedCounter.Inc()
 			return nil, err
 		}
-		m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
+
+		// m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req)
 		prommetrics.SingleChatMsgProcessSuccessCounter.Inc()
 		return &pbmsg.SendMsgResp{
 			ServerMsgID: req.MsgData.ServerMsgID,
@@ -201,3 +203,25 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 		}, nil
 	}
 }
+
+func (m *msgServer) SendSimpleMsg(ctx context.Context, req *pbmsg.SendSimpleMsgReq) (*pbmsg.SendSimpleMsgResp, error) {
+	if req.MsgData == nil {
+		return nil, errs.ErrArgs.WrapMsg("msg data is nil")
+	}
+	sender, err := m.UserLocalCache.GetUserInfo(ctx, req.MsgData.SendID)
+	if err != nil {
+		return nil, err
+	}
+	req.MsgData.SenderFaceURL = sender.FaceURL
+	req.MsgData.SenderNickname = sender.Nickname
+	resp, err := m.SendMsg(ctx, &pbmsg.SendMsgReq{MsgData: req.MsgData})
+	if err != nil {
+		return nil, err
+	}
+	return &pbmsg.SendSimpleMsgResp{
+		ServerMsgID: resp.ServerMsgID,
+		ClientMsgID: resp.ClientMsgID,
+		SendTime:    resp.SendTime,
+		Modify:      resp.Modify,
+	}, nil
+}
@@ -22,6 +22,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/dbbuild"
 	"github.com/openimsdk/open-im-server/v3/pkg/mqbuild"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+	"google.golang.org/grpc"

 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
@@ -35,7 +36,6 @@ import (
 	"github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/discovery"
-	"google.golang.org/grpc"
 )

 type MessageInterceptorFunc func(ctx context.Context, globalConfig *Config, req *msg.SendMsgReq) (*sdkws.MsgData, error)
@@ -70,6 +70,8 @@ type msgServer struct {
 	config                 *Config                          // Global configuration settings.
 	webhookClient          *webhook.Client
 	conversationClient     *rpcli.ConversationClient
+
+	adminUserIDs []string
 }

 func (m *msgServer) addInterceptorHandler(interceptorFunc ...MessageInterceptorFunc) {
@@ -77,7 +79,7 @@ func (m *msgServer) addInterceptorHandler(interceptorFunc ...MessageInterceptorF

 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	builder := mqbuild.NewBuilder(&config.KafkaConfig)
 	redisProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToRedisTopic)
 	if err != nil {
@@ -144,6 +146,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 		config:                 config,
 		webhookClient:          webhook.NewWebhookClient(config.WebhooksConfig.URL),
 		conversationClient:     conversationClient,
+		adminUserIDs:           config.Share.IMAdminUser.UserIDs,
 	}

 	s.notificationSender = notification.NewNotificationSender(&config.NotificationConfig, notification.WithLocalSendMsg(s.SendMsg))
@@ -54,7 +54,7 @@ type MessageRevoked struct {
 func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgReq) error {
 	switch data.MsgData.SessionType {
 	case constant.SingleChatType:
-		if datautil.Contain(data.MsgData.SendID, m.config.Share.IMAdminUserID...) {
+		if datautil.Contain(data.MsgData.SendID, m.adminUserIDs...) {
 			return nil
 		}
 		if data.MsgData.ContentType <= constant.NotificationEnd &&
@@ -102,7 +102,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			return nil
 		}

-		if datautil.Contain(data.MsgData.SendID, m.config.Share.IMAdminUserID...) {
+		if datautil.Contain(data.MsgData.SendID, m.adminUserIDs...) {
 			return nil
 		}
 		if data.MsgData.ContentType <= constant.NotificationEnd &&
@@ -18,10 +18,9 @@ import (
 	"context"
 	"time"

-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/errs"
@@ -66,7 +66,7 @@ type Config struct {
 	Discovery          config.Discovery
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig)
 	mgocli, err := dbb.Mongo(ctx)
 	if err != nil {
@@ -192,7 +192,7 @@ func (s *friendServer) ImportFriends(ctx context.Context, req *relation.ImportFr
 			FromUserID:   req.OwnerUserID,
 			ToUserID:     userID,
 			HandleResult: constant.FriendResponseAgree,
-		})
+		}, false)
 	}

 	s.webhookAfterImportFriends(ctx, &s.config.WebhooksConfig.AfterImportFriends, req)
@@ -221,7 +221,7 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *relation.Res
 			return nil, err
 		}
 		s.webhookAfterAddFriendAgree(ctx, &s.config.WebhooksConfig.AfterAddFriendAgree, req)
-		s.notificationSender.FriendApplicationAgreedNotification(ctx, req)
+		s.notificationSender.FriendApplicationAgreedNotification(ctx, req, true)
 		return resp, nil
 	}
 	if req.HandleResult == constant.FriendResponseRefuse {
@@ -171,11 +171,17 @@ func (f *FriendNotificationSender) FriendApplicationAddNotification(ctx context.
 	f.Notification(ctx, req.FromUserID, req.ToUserID, constant.FriendApplicationNotification, &tips)
 }

-func (f *FriendNotificationSender) FriendApplicationAgreedNotification(ctx context.Context, req *relation.RespondFriendApplyReq) {
-	request, err := f.getFriendRequests(ctx, req.FromUserID, req.ToUserID)
-	if err != nil {
-		log.ZError(ctx, "FriendApplicationAgreedNotification get friend request", err, "fromUserID", req.FromUserID, "toUserID", req.ToUserID)
-		return
+func (f *FriendNotificationSender) FriendApplicationAgreedNotification(ctx context.Context, req *relation.RespondFriendApplyReq, checkReq bool) {
+	var (
+		request *sdkws.FriendRequest
+		err     error
+	)
+	if checkReq {
+		request, err = f.getFriendRequests(ctx, req.FromUserID, req.ToUserID)
+		if err != nil {
+			log.ZError(ctx, "FriendApplicationAgreedNotification get friend request", err, "fromUserID", req.FromUserID, "toUserID", req.ToUserID)
+			return
+		}
 	}
 	tips := sdkws.FriendApplicationApprovedTips{
 		FromToUserID: &sdkws.FromToUserID{
@@ -64,7 +64,7 @@ type Config struct {
 	Discovery          config.Discovery
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig)
 	mgocli, err := dbb.Mongo(ctx)
 	if err != nil {
@@ -65,6 +65,8 @@ type userServer struct {
 	groupClient              *rpcli.GroupClient
 	relationClient           *rpcli.RelationClient
 	clientConfig             controller.ClientConfigDatabase
+
+	adminUserIDs []string
 }

 type Config struct {
@@ -79,7 +81,7 @@ type Config struct {
 	Discovery          config.Discovery
 }

-func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
 	dbb := dbbuild.NewBuilder(&config.MongodbConfig, &config.RedisConfig)
 	mgocli, err := dbb.Mongo(ctx)
 	if err != nil {
@@ -92,8 +94,12 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr

 	users := make([]*tablerelation.User, 0)

-	for _, v := range config.Share.IMAdminUserID {
-		users = append(users, &tablerelation.User{UserID: v, Nickname: v, AppMangerLevel: constant.AppAdmin})
+	for i := range config.Share.IMAdminUser.UserIDs {
+		users = append(users, &tablerelation.User{
+			UserID:         config.Share.IMAdminUser.UserIDs[i],
+			Nickname:       config.Share.IMAdminUser.Nicknames[i],
+			AppMangerLevel: constant.AppAdmin,
+		})
 	}
 	userDB, err := mgo.NewUserMongo(mgocli.GetDB())
 	if err != nil {
@@ -130,6 +136,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 		clientConfig:             controller.NewClientConfigDatabase(clientConfigDB, redis.NewClientConfigCache(rdb, clientConfigDB), mgocli.GetTx()),
 		groupClient:              rpcli.NewGroupClient(groupConn),
 		relationClient:           rpcli.NewRelationClient(friendConn),
+		adminUserIDs:             config.Share.IMAdminUser.UserIDs,
 	}
 	pbuser.RegisterUserServer(server, u)
 	return u.db.InitOnce(context.Background(), users)
@@ -197,6 +204,7 @@ func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUse
 	}

 	s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID)
+
 	//friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID)
 	//if err != nil {
 	//	return nil, err
@@ -209,6 +217,7 @@ func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUse
 	//for _, friendID := range friends {
 	//	s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
 	//}
+
 	s.webhookAfterUpdateUserInfoEx(ctx, &s.config.WebhooksConfig.AfterUpdateUserInfoEx, req)
 	if err := s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID, oldUser); err != nil {
 		return nil, err
@@ -646,7 +655,7 @@ func (s *userServer) userModelToResp(users []*tablerelation.User, pagination pag
 	accounts := make([]*pbuser.NotificationAccountInfo, 0)
 	var total int64
 	for _, v := range users {
-		if v.AppMangerLevel >= constant.AppNotificationAdmin && !datautil.Contain(v.UserID, s.config.Share.IMAdminUserID...) {
+		if v.AppMangerLevel >= constant.AppNotificationAdmin && !datautil.Contain(v.UserID, s.adminUserIDs...) {
 			if appManagerLevel != nil {
 				if v.AppMangerLevel != *appManagerLevel {
 					continue
@@ -10,7 +10,6 @@ import (
 	"github.com/openimsdk/protocol/third"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
-
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
@@ -25,14 +24,14 @@ type Config struct {
 	Discovery config.Discovery
 }

-func Start(ctx context.Context, conf *Config, client discovery.Conn, service grpc.ServiceRegistrar) error {
+func Start(ctx context.Context, conf *Config, client discovery.SvcDiscoveryRegistry, service grpc.ServiceRegistrar) error {
 	log.CInfo(ctx, "CRON-TASK server is initializing", "runTimeEnv", runtimeenv.RuntimeEnvironment(), "chatRecordsClearTime", conf.CronTask.CronExecuteTime, "msgDestructTime", conf.CronTask.RetainChatRecords)
 	if conf.CronTask.RetainChatRecords < 1 {
 		log.ZInfo(ctx, "disable cron")
 		<-ctx.Done()
 		return nil
 	}
-	ctx = mcontext.SetOpUserID(ctx, conf.Share.IMAdminUserID[0])
+	ctx = mcontext.SetOpUserID(ctx, conf.Share.IMAdminUser.UserIDs[0])

 	msgConn, err := client.GetConn(ctx, conf.Discovery.RpcService.Msg)
 	if err != nil {
@@ -49,6 +48,7 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 		return err
 	}

+	var locker Locker
 	if conf.Discovery.Enable == config.ETCD {
 		cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), []string{
 			conf.CronTask.GetConfigFileName(),
@@ -56,6 +56,14 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 			conf.Discovery.GetConfigFileName(),
 		})
 		cm.Watch(ctx)
+		locker, err = NewEtcdLocker(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient())
+		if err != nil {
+			return err
+		}
+	}
+
+	if locker == nil {
+		locker = emptyLocker{}
 	}

 	srv := &cronServer{
@@ -65,6 +73,7 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 		msgClient:          msg.NewMsgClient(msgConn),
 		conversationClient: pbconversation.NewConversationClient(conversationConn),
 		thirdClient:        third.NewThirdClient(thirdConn),
+		locker:             locker,
 	}

 	if err := srv.registerClearS3(); err != nil {
@@ -81,9 +90,21 @@ func Start(ctx context.Context, conf *Config, client discovery.Conn, service grp
 	log.ZDebug(ctx, "cron task server is running")
 	<-ctx.Done()
 	log.ZDebug(ctx, "cron task server is shutting down")
+	srv.cron.Stop()
+
 	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
@@ -91,6 +112,7 @@ type cronServer struct {
 	msgClient          msg.MsgClient
 	conversationClient pbconversation.ConversationClient
 	thirdClient        third.ThirdClient
+	locker             Locker
 }

 func (c *cronServer) registerClearS3() error {
@@ -98,7 +120,9 @@ func (c *cronServer) registerClearS3() error {
 		log.ZInfo(c.ctx, "disable scheduled cleanup of s3", "fileExpireTime", c.config.CronTask.FileExpireTime, "deleteObjectType", c.config.CronTask.DeleteObjectType)
 		return nil
 	}
-	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.clearS3)
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
+		c.locker.ExecuteWithLock(c.ctx, "clearS3", c.clearS3)
+	})
 	return errs.WrapMsg(err, "failed to register clear s3 cron task")
 }

@@ -107,11 +131,15 @@ func (c *cronServer) registerDeleteMsg() error {
 		log.ZInfo(c.ctx, "disable scheduled cleanup of chat records", "retainChatRecords", c.config.CronTask.RetainChatRecords)
 		return nil
 	}
-	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.deleteMsg)
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
+		c.locker.ExecuteWithLock(c.ctx, "deleteMsg", c.deleteMsg)
+	})
 	return errs.WrapMsg(err, "failed to register delete msg cron task")
 }

 func (c *cronServer) registerClearUserMsg() error {
-	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.clearUserMsg)
+	_, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, func() {
+		c.locker.ExecuteWithLock(c.ctx, "clearUserMsg", c.clearUserMsg)
+	})
 	return errs.WrapMsg(err, "failed to register clear user msg cron task")
 }
@@ -0,0 +1,86 @@
+package cron
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/openimsdk/tools/log"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/concurrency"
+)
+
+const (
+	lockLeaseTTL = 300
+)
+
+type EtcdLocker struct {
+	client     *clientv3.Client
+	instanceID string
+}
+
+// NewEtcdLocker creates a new etcd distributed lock
+func NewEtcdLocker(client *clientv3.Client) (*EtcdLocker, error) {
+	hostname, _ := os.Hostname()
+	pid := os.Getpid()
+	instanceID := fmt.Sprintf("%s-pid-%d-%d", hostname, pid, time.Now().UnixNano())
+
+	locker := &EtcdLocker{
+		client:     client,
+		instanceID: instanceID,
+	}
+
+	return locker, nil
+}
+
+func (e *EtcdLocker) ExecuteWithLock(ctx context.Context, taskName string, task func()) {
+	session, err := concurrency.NewSession(e.client, concurrency.WithTTL(lockLeaseTTL))
+	if err != nil {
+		log.ZWarn(ctx, "Failed to create etcd session", err,
+			"taskName", taskName,
+			"instanceID", e.instanceID)
+		return
+	}
+	defer session.Close()
+
+	lockKey := fmt.Sprintf("openim/crontask/%s", taskName)
+	mutex := concurrency.NewMutex(session, lockKey)
+
+	ctxWithTimeout, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
+	defer cancel()
+
+	err = mutex.TryLock(ctxWithTimeout)
+	if err != nil {
+		// errors.Is(err, concurrency.ErrLocked)
+		log.ZDebug(ctx, "Task is being executed by another instance, skipping",
+			"taskName", taskName,
+			"instanceID", e.instanceID,
+			"error", err.Error())
+
+		return
+	}
+
+	defer func() {
+		if err := mutex.Unlock(ctx); err != nil {
+			log.ZWarn(ctx, "Failed to release task lock", err,
+				"taskName", taskName,
+				"instanceID", e.instanceID)
+		} else {
+			log.ZInfo(ctx, "Successfully released task lock",
+				"taskName", taskName,
+				"instanceID", e.instanceID)
+		}
+	}()
+
+	log.ZInfo(ctx, "Successfully acquired task lock, starting execution",
+		"taskName", taskName,
+		"instanceID", e.instanceID,
+		"sessionID", session.Lease())
+
+	task()
+
+	log.ZInfo(ctx, "Task execution completed",
+		"taskName", taskName,
+		"instanceID", e.instanceID)
+}