merge: merge main code into js branch. (#2648)

* feat: update group notification when set to null. (#2590)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* feat: update group notification when set to null.

* update log standard.

* feat: add long time push msg in prometheus (#2584)

* feat: add long time push msg in prometheus

* fix: log print

* fix: go mod

* fix: log msg

* fix: log init

* feat: push msg

* feat: go mod ,remove cgo package

* feat: remove error log

* feat: test dummy push

* feat:redis pool config

* feat: push to kafka log

* feat: supports getting messages based on session ID and seq (#2582)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* feat: implement request batch count limit. (#2591)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* fix: getting messages based on session ID and seq (#2595)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* feat: avoid pulling messages from sessions with a large number of max seq values of 0 (#2602)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* refactor: improve db structure in `storage/controller` (#2604)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* refactor: improve db structure in `storage/controller`

* feat: implement offline push using kafka (#2600)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* feat: implement offline push.

* feat: implement batch Push spilt

* update go mod

* feat: implement kafka producer and consumer.

* update format,

* add PushMQ log.

* feat: update Handler logic.

* update MQ logic.

* update

* update

* fix: update OfflinePushConsumerHandler.

* feat: API supports gzip (#2609)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* Fix err (#2608)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* feat: add rocksTimeout

* feat: wrap logs

* feat: add logs

* feat: listen config

* feat: enable listen TIME_WAIT port

* feat: add logs

* feat: cache batch

* chore: enable fullUserCache

* feat: push rpc num

* feat: push err

* feat: with operationID

* feat: sleep

* feat: change 1s

* feat: change log

* feat: implement Getbatch in rpcCache.

* feat: print getOnline cost

* feat: change log

* feat: change kafka and push config

* feat: del interface

* feat: fix err

* feat: change config

* feat: go mod

* feat: change config

* feat: change config

* feat: add sleep in push

* feat: warn logs

* feat: logs

* feat: logs

* feat: change port

* feat: start config

* feat: remove port reuse

* feat: prometheus config

* feat: prometheus config

* feat: prometheus config

* feat: add long time send msg to grafana

* feat: init

* feat: init

* feat: implement offline push.

* feat: batch get user online

* feat: implement batch Push spilt

* update go mod

* Revert "feat: change port"

This reverts commit 06d5e944

* feat: change port

* feat: change config

* feat: implement kafka producer and consumer.

* update format,

* add PushMQ log.

* feat: get all online users and init push

* feat: lock in online cache

* feat: config

* fix: init online status

* fix: add logs

* fix: userIDs

* fix: add logs

* feat: update Handler logic.

* update MQ logic.

* update

* update

* fix: method name

* fix: update OfflinePushConsumerHandler.

* fix: prommetrics

* fix: add logs

* fix: ctx

* fix: log

* fix: config

* feat: change port

* fix: atomic online cache status

---------

Co-authored-by: Monet Lee <monet_lee@163.com>

* feature: add GetConversationsHasReadAndMaxSeq interface to the WebSocket API. (#2611)

* fix: lru lock (#2613)

* fix: lru lock

* fix: lru lock

* fix: lru lock

* fix: nil pointer error on close (#2618)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

* go.mod

* fix: nil pointer error on close

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* feat: create group can push notification (#2617)

* fix: blockage caused by listen error (#2620)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

* go.mod

* fix: nil pointer error on close

* fix: listen error

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* fix: go.mod (#2621)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

* go.mod

* fix: nil pointer error on close

* fix: listen error

* fix: listen error

* update go.mod

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* feat: improve searchMsg implement. (#2614)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* remove unused script.

* feat: improve searchMsg implement.

* update mongo config.

* Fix lock (#2622)

* fix:log

* fix: lock

* fix: update setGroupInfoEX field name. (#2625)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* fix: update setGroupInfoEX field name.

* fix: update setGroupInfoEX field name (#2626)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* fix: update setGroupInfoEX field name.

* fix: update setGroupInfoEX field name

* feat: msg gateway add log (#2631)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

* go.mod

* fix: nil pointer error on close

* fix: listen error

* fix: listen error

* update go.mod

* feat: add log

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* fix: update setGroupInfoEx func name and field. (#2634)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* fix: update setGroupInfoEx func name and field.

* refactor: update groupinfoEx field.

* refactor: update database name in mongodb.yml

* add groupName Condition

* fix: fix setConversations req fill. (#2645)

* refactor: refactor workflows contents.

* add tool workflows.

* update field.

* fix: remove chat error.

* Fix err.

* fix error.

* remove cn comment.

* update workflows files.

* update infra config.

* move workflows.

* feat: update bot.

* fix: solve uncorrect outdated msg get.

* update get docIDs logic.

* update

* update skip logic.

* fix

* update.

* fix: delay deleteObject func.

* remove unused content.

* update log type.

* feat: implement request batch count limit.

* update

* update

* fix: fix setConversations req fill.

* fix: GetMsgBySeqs boundary issues (#2647)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* update gomake version

* update gomake version

* fix: seq conversion bug

* fix: redis pipe exec

* fix: ImportFriends

* fix: A large number of logs keysAndValues ​​length is not even

* feat: mark read aggregate write

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* feat: online status supports redis cluster

* merge

* merge

* read seq is written to mongo

* read seq is written to mongo

* fix: invitation to join group notification

* fix: friend op_user_id

* feat: optimizing asynchronous context

* feat: optimizing memamq size

* feat: add GetSeqMessage

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: GroupApplicationAgreeMemberEnterNotification

* feat: go.mod

* feat: go.mod

* feat: join group notification and get seq

* feat: join group notification and get seq

* feat: avoid pulling messages from sessions with a large number of max seq values of 0

* feat: API supports gzip

* go.mod

* fix: nil pointer error on close

* fix: listen error

* fix: listen error

* update go.mod

* feat: add log

* fix: token parse token value

* fix: GetMsgBySeqs boundary issues

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

* fix: the attribute version is obsolete, remove it (#2644)

* refactor: update Userregister request field. (#2650)

---------

Co-authored-by: Monet Lee <monet_lee@163.com>
Co-authored-by: icey-yu <119291641+icey-yu@users.noreply.github.com>
Co-authored-by: chao <48119764+withchao@users.noreply.github.com>
Co-authored-by: withchao <withchao@users.noreply.github.com>
Co-authored-by: 蔡相跃 <caixiangyue007@gmail.com>

pkg/callbackstruct/constant.go

@@ -18,9 +18,9 @@ const (
 	CallbackBeforeInviteJoinGroupCommand    = "callbackBeforeInviteJoinGroupCommand"
 	CallbackAfterJoinGroupCommand           = "callbackAfterJoinGroupCommand"
 	CallbackAfterSetGroupInfoCommand        = "callbackAfterSetGroupInfoCommand"
-	CallbackAfterSetGroupInfoEXCommand      = "callbackAfterSetGroupInfoCommandEX"
+	CallbackAfterSetGroupInfoExCommand      = "callbackAfterSetGroupInfoExCommand"
 	CallbackBeforeSetGroupInfoCommand       = "callbackBeforeSetGroupInfoCommand"
-	CallbackBeforeSetGroupInfoEXCommand     = "callbackBeforeSetGroupInfoEXCommand"
+	CallbackBeforeSetGroupInfoExCommand     = "callbackBeforeSetGroupInfoExCommand"
 	CallbackAfterRevokeMsgCommand           = "callbackBeforeAfterMsgCommand"
 	CallbackBeforeAddBlackCommand           = "callbackBeforeAddBlackCommand"
 	CallbackAfterAddFriendCommand           = "callbackAfterAddFriendCommand"

pkg/callbackstruct/group.go

@@ -244,11 +244,11 @@ type CallbackAfterSetGroupInfoResp struct {
 	CommonCallbackResp
 }
 
-type CallbackBeforeSetGroupInfoEXReq struct {
+type CallbackBeforeSetGroupInfoExReq struct {
 	CallbackCommand   `json:"callbackCommand"`
 	OperationID       string                  `json:"operationID"`
 	GroupID           string                  `json:"groupID"`
-	GroupName         string                  `json:"groupName"`
+	GroupName         *wrapperspb.StringValue `json:"groupName"`
 	Notification      *wrapperspb.StringValue `json:"notification"`
 	Introduction      *wrapperspb.StringValue `json:"introduction"`
 	FaceURL           *wrapperspb.StringValue `json:"faceURL"`
@@ -258,10 +258,10 @@ type CallbackBeforeSetGroupInfoEXReq struct {
 	ApplyMemberFriend *wrapperspb.Int32Value  `json:"applyMemberFriend"`
 }
 
-type CallbackBeforeSetGroupInfoEXResp struct {
+type CallbackBeforeSetGroupInfoExResp struct {
 	CommonCallbackResp
 	GroupID           string                  `json:"groupID"`
-	GroupName         string                  `json:"groupName"`
+	GroupName         *wrapperspb.StringValue `json:"groupName"`
 	Notification      *wrapperspb.StringValue `json:"notification"`
 	Introduction      *wrapperspb.StringValue `json:"introduction"`
 	FaceURL           *wrapperspb.StringValue `json:"faceURL"`
@@ -271,11 +271,11 @@ type CallbackBeforeSetGroupInfoEXResp struct {
 	ApplyMemberFriend *wrapperspb.Int32Value  `json:"applyMemberFriend"`
 }
 
-type CallbackAfterSetGroupInfoEXReq struct {
+type CallbackAfterSetGroupInfoExReq struct {
 	CallbackCommand   `json:"callbackCommand"`
 	OperationID       string                  `json:"operationID"`
 	GroupID           string                  `json:"groupID"`
-	GroupName         string                  `json:"groupName"`
+	GroupName         *wrapperspb.StringValue `json:"groupName"`
 	Notification      *wrapperspb.StringValue `json:"notification"`
 	Introduction      *wrapperspb.StringValue `json:"introduction"`
 	FaceURL           *wrapperspb.StringValue `json:"faceURL"`
@@ -285,6 +285,6 @@ type CallbackAfterSetGroupInfoEXReq struct {
 	ApplyMemberFriend *wrapperspb.Int32Value  `json:"applyMemberFriend"`
 }
 
-type CallbackAfterSetGroupInfoEXResp struct {
+type CallbackAfterSetGroupInfoExResp struct {
 	CommonCallbackResp
 }

pkg/callbackstruct/user.go

@@ -72,7 +72,6 @@ type CallbackAfterUpdateUserInfoExResp struct {
 
 type CallbackBeforeUserRegisterReq struct {
 	CallbackCommand `json:"callbackCommand"`
-	Secret          string            `json:"secret"`
 	Users           []*sdkws.UserInfo `json:"users"`
 }
 
@@ -83,7 +82,6 @@ type CallbackBeforeUserRegisterResp struct {
 
 type CallbackAfterUserRegisterReq struct {
 	CallbackCommand `json:"callbackCommand"`
-	Secret          string            `json:"secret"`
 	Users           []*sdkws.UserInfo `json:"users"`
 }
 

pkg/common/cmd/push.go

@@ -37,7 +37,6 @@ func NewPushRpcCmd() *PushRpcCmd {
 	ret.configMap = map[string]any{
 		OpenIMPushCfgFileName:    &pushConfig.RpcConfig,
 		RedisConfigFileName:      &pushConfig.RedisConfig,
-		MongodbConfigFileName:    &pushConfig.MongodbConfig,
 		KafkaConfigFileName:      &pushConfig.KafkaConfig,
 		ShareFileName:            &pushConfig.Share,
 		NotificationFileName:     &pushConfig.NotificationConfig,

pkg/common/cmd/root.go

@@ -129,10 +129,11 @@ func (r *RootCmd) applyOptions(opts ...func(*CmdOpts)) *CmdOpts {
 }
 
 func (r *RootCmd) initializeLogger(cmdOpts *CmdOpts) error {
-	err := log.InitFromConfig(
+	err := log.InitLoggerFromConfig(
 
 		cmdOpts.loggerPrefixName,
 		r.processName,
+		"", "",
 		r.log.RemainLogLevel,
 		r.log.IsStdout,
 		r.log.IsJson,

pkg/common/config/config.go

@@ -73,18 +73,21 @@ type Mongo struct {
 	MaxRetry    int      `mapstructure:"maxRetry"`
 }
 type Kafka struct {
-	Username       string    `mapstructure:"username"`
-	Password       string    `mapstructure:"password"`
-	ProducerAck    string    `mapstructure:"producerAck"`
-	CompressType   string    `mapstructure:"compressType"`
-	Address        []string  `mapstructure:"address"`
-	ToRedisTopic   string    `mapstructure:"toRedisTopic"`
-	ToMongoTopic   string    `mapstructure:"toMongoTopic"`
-	ToPushTopic    string    `mapstructure:"toPushTopic"`
-	ToRedisGroupID string    `mapstructure:"toRedisGroupID"`
-	ToMongoGroupID string    `mapstructure:"toMongoGroupID"`
-	ToPushGroupID  string    `mapstructure:"toPushGroupID"`
-	Tls            TLSConfig `mapstructure:"tls"`
+	Username           string   `mapstructure:"username"`
+	Password           string   `mapstructure:"password"`
+	ProducerAck        string   `mapstructure:"producerAck"`
+	CompressType       string   `mapstructure:"compressType"`
+	Address            []string `mapstructure:"address"`
+	ToRedisTopic       string   `mapstructure:"toRedisTopic"`
+	ToMongoTopic       string   `mapstructure:"toMongoTopic"`
+	ToPushTopic        string   `mapstructure:"toPushTopic"`
+	ToOfflinePushTopic string   `mapstructure:"toOfflinePushTopic"`
+	ToRedisGroupID     string   `mapstructure:"toRedisGroupID"`
+	ToMongoGroupID     string   `mapstructure:"toMongoGroupID"`
+	ToPushGroupID      string   `mapstructure:"toPushGroupID"`
+	ToOfflineGroupID   string   `mapstructure:"toOfflinePushGroupID"`
+
+	Tls TLSConfig `mapstructure:"tls"`
 }
 type TLSConfig struct {
 	EnableTLS          bool   `mapstructure:"enableTLS"`
@@ -97,8 +100,9 @@ type TLSConfig struct {
 
 type API struct {
 	Api struct {
-		ListenIP string `mapstructure:"listenIP"`
-		Ports    []int  `mapstructure:"ports"`
+		ListenIP         string `mapstructure:"listenIP"`
+		Ports            []int  `mapstructure:"ports"`
+		CompressionLevel int    `mapstructure:"compressionLevel"`
 	} `mapstructure:"api"`
 	Prometheus struct {
 		Enable     bool   `mapstructure:"enable"`
@@ -220,6 +224,7 @@ type Push struct {
 		BadgeCount bool   `mapstructure:"badgeCount"`
 		Production bool   `mapstructure:"production"`
 	} `mapstructure:"iosPush"`
+	FullUserCache bool `mapstructure:"fullUserCache"`
 }
 
 type Auth struct {
@@ -336,7 +341,8 @@ type Redis struct {
 	Password    string   `mapstructure:"password"`
 	ClusterMode bool     `mapstructure:"clusterMode"`
 	DB          int      `mapstructure:"storage"`
-	MaxRetry    int      `mapstructure:"MaxRetry"`
+	MaxRetry    int      `mapstructure:"maxRetry"`
+	PoolSize    int      `mapstructure:"poolSize"`
 }
 
 type BeforeConfig struct {
@@ -422,8 +428,8 @@ type Webhooks struct {
 	BeforeInviteUserToGroup  BeforeConfig `mapstructure:"beforeInviteUserToGroup"`
 	AfterSetGroupInfo        AfterConfig  `mapstructure:"afterSetGroupInfo"`
 	BeforeSetGroupInfo       BeforeConfig `mapstructure:"beforeSetGroupInfo"`
-	AfterSetGroupInfoEX      AfterConfig  `mapstructure:"afterSetGroupInfoEX"`
-	BeforeSetGroupInfoEX     BeforeConfig `mapstructure:"beforeSetGroupInfoEX"`
+	AfterSetGroupInfoEx      AfterConfig  `mapstructure:"afterSetGroupInfoEx"`
+	BeforeSetGroupInfoEx     BeforeConfig `mapstructure:"beforeSetGroupInfoEx"`
 	AfterRevokeMsg           AfterConfig  `mapstructure:"afterRevokeMsg"`
 	BeforeAddBlack           BeforeConfig `mapstructure:"beforeAddBlack"`
 	AfterAddFriend           AfterConfig  `mapstructure:"afterAddFriend"`
@@ -474,6 +480,7 @@ func (r *Redis) Build() *redisutil.Config {
 		Password:    r.Password,
 		DB:          r.DB,
 		MaxRetry:    r.MaxRetry,
+		PoolSize:    r.PoolSize,
 	}
 }
 

pkg/common/prommetrics/grpc_push.go

@@ -23,4 +23,8 @@ var (
 		Name: "msg_offline_push_failed_total",
 		Help: "The number of msg failed offline pushed",
 	})
+	MsgLoneTimePushCounter = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "msg_long_time_push_total",
+		Help: "The number of messages with a push time exceeding 10 seconds",
+	})
 )

pkg/common/prommetrics/rpc.go

@@ -47,9 +47,17 @@ func GetGrpcCusMetrics(registerName string, share *config.Share) []prometheus.Co
 	case share.RpcRegisterName.MessageGateway:
 		return []prometheus.Collector{OnlineUserGauge}
 	case share.RpcRegisterName.Msg:
-		return []prometheus.Collector{SingleChatMsgProcessSuccessCounter, SingleChatMsgProcessFailedCounter, GroupChatMsgProcessSuccessCounter, GroupChatMsgProcessFailedCounter}
+		return []prometheus.Collector{
+			SingleChatMsgProcessSuccessCounter,
+			SingleChatMsgProcessFailedCounter,
+			GroupChatMsgProcessSuccessCounter,
+			GroupChatMsgProcessFailedCounter,
+		}
 	case share.RpcRegisterName.Push:
-		return []prometheus.Collector{MsgOfflinePushFailedCounter}
+		return []prometheus.Collector{
+			MsgOfflinePushFailedCounter,
+			MsgLoneTimePushCounter,
+		}
 	case share.RpcRegisterName.Auth:
 		return []prometheus.Collector{UserLoginCounter}
 	case share.RpcRegisterName.User:

pkg/common/startrpc/start.go

@@ -25,7 +25,6 @@ import (
 	"os"
 	"os/signal"
 	"strconv"
-	"sync"
 	"syscall"
 	"time"
 
@@ -35,7 +34,6 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mw"
-	"github.com/openimsdk/tools/system/program"
 	"github.com/openimsdk/tools/utils/network"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -54,6 +52,7 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 	log.CInfo(ctx, "RPC server is initializing", "rpcRegisterName", rpcRegisterName, "rpcPort", rpcPort,
 		"prometheusPorts", prometheusConfig.Ports)
 	rpcTcpAddr := net.JoinHostPort(network.GetListenIP(listenIP), strconv.Itoa(rpcPort))
+
 	listener, err := net.Listen(
 		"tcp",
 		rpcTcpAddr,
@@ -61,7 +60,6 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 	if err != nil {
 		return errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
 	}
-
 	defer listener.Close()
 	client, err := kdisc.NewDiscoveryRegister(discovery, share)
 	if err != nil {
@@ -92,10 +90,6 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 	}
 
 	srv := grpc.NewServer(options...)
-	once := sync.Once{}
-	defer func() {
-		once.Do(srv.GracefulStop)
-	}()
 
 	err = rpcFn(ctx, config, client, srv)
 	if err != nil {
@@ -113,9 +107,8 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 	}
 
 	var (
-		netDone    = make(chan struct{}, 2)
-		netErr     error
-		httpServer *http.Server
+		netDone = make(chan struct{}, 2)
+		netErr  error
 	)
 	if prometheusConfig.Enable {
 		go func() {
@@ -152,18 +145,11 @@ func Start[T any](ctx context.Context, discovery *config.Discovery, prometheusCo
 	signal.Notify(sigs, syscall.SIGTERM)
 	select {
 	case <-sigs:
-		program.SIGTERMExit()
-		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
 		if err := gracefulStopWithCtx(ctx, srv.GracefulStop); err != nil {
 			return err
 		}
-		ctx, cancel = context.WithTimeout(context.Background(), 15*time.Second)
-		defer cancel()
-		err := httpServer.Shutdown(ctx)
-		if err != nil {
-			return errs.WrapMsg(err, "shutdown err")
-		}
 		return nil
 	case <-netDone:
 		return netErr

pkg/common/storage/cache/cachekey/online.go

@@ -1,6 +1,9 @@
 package cachekey
 
-import "time"
+import (
+	"strings"
+	"time"
+)
 
 const (
 	OnlineKey     = "ONLINE:"
@@ -11,3 +14,7 @@ const (
 func GetOnlineKey(userID string) string {
 	return OnlineKey + userID
 }
+
+func GetOnlineKeyUserID(key string) string {
+	return strings.TrimPrefix(key, OnlineKey)
+}

pkg/common/storage/cache/online.go

@@ -5,4 +5,5 @@ import "context"
 type OnlineCache interface {
 	GetOnline(ctx context.Context, userID string) ([]int32, error)
 	SetUserOnline(ctx context.Context, userID string, online, offline []int32) error
+	GetAllOnlineUsers(ctx context.Context, cursor uint64) (map[string][]int32, uint64, error)
 }

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"github.com/dtm-labs/rockscache"
+	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/redis/go-redis/v9"
 	"golang.org/x/sync/singleflight"
@@ -65,6 +66,7 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rockscac
 				}
 				bs, err := json.Marshal(value)
 				if err != nil {
+					log.ZError(ctx, "marshal failed", err)
 					return nil, err
 				}
 				cacheIndex[index] = string(bs)
@@ -72,7 +74,7 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rockscac
 			return cacheIndex, nil
 		})
 		if err != nil {
-			return nil, err
+			return nil, errs.WrapMsg(err, "FetchBatch2 failed")
 		}
 		for index, data := range indexCache {
 			if data == "" {
@@ -80,7 +82,7 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rockscac
 			}
 			var value V
 			if err := json.Unmarshal([]byte(data), &value); err != nil {
-				return nil, err
+				return nil, errs.WrapMsg(err, "Unmarshal failed")
 			}
 			if cb, ok := any(&value).(BatchCacheCallback[K]); ok {
 				cb.BatchCache(keyId[keys[index]])

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

@@ -28,6 +28,10 @@ import (
 	"time"
 )
 
+const (
+	rocksCacheTimeout = 11 * time.Second
+)
+
 // BatchDeleterRedis is a concrete implementation of the BatchDeleter interface based on Redis and RocksCache.
 type BatchDeleterRedis struct {
 	redisClient    redis.UniversalClient
@@ -106,6 +110,8 @@ func (c *BatchDeleterRedis) AddKeys(keys ...string) {
 // GetRocksCacheOptions returns the default configuration options for RocksCache.
 func GetRocksCacheOptions() *rockscache.Options {
 	opts := rockscache.NewDefaultOptions()
+	opts.LockExpire = rocksCacheTimeout
+	opts.WaitReplicasTimeout = rocksCacheTimeout
 	opts.StrongConsistency = true
 	opts.RandomExpireAdjustment = 0.2
 
@@ -118,7 +124,7 @@ func getCache[T any](ctx context.Context, rcClient *rockscache.Client, key strin
 	v, err := rcClient.Fetch2(ctx, key, expire, func() (s string, err error) {
 		t, err = fn(ctx)
 		if err != nil {
-			log.ZError(ctx, "getCache query database failed", err, "key", key)
+			//log.ZError(ctx, "getCache query database failed", err, "key", key)
 			return "", err
 		}
 		bs, err := json.Marshal(t)

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

@@ -2,8 +2,10 @@ package redis
 
 import (
 	"context"
+	"fmt"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
+	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/redis/go-redis/v9"
@@ -49,6 +51,36 @@ func (s *userOnline) GetOnline(ctx context.Context, userID string) ([]int32, err
 	return platformIDs, nil
 }
 
+func (s *userOnline) GetAllOnlineUsers(ctx context.Context, cursor uint64) (map[string][]int32, uint64, error) {
+	result := make(map[string][]int32)
+
+	keys, nextCursor, err := s.rdb.Scan(ctx, cursor, fmt.Sprintf("%s*", cachekey.OnlineKey), constant.ParamMaxLength).Result()
+	if err != nil {
+		return nil, 0, err
+	}
+
+	for _, key := range keys {
+		userID := cachekey.GetOnlineKeyUserID(key)
+		strValues, err := s.rdb.ZRange(ctx, key, 0, -1).Result()
+		if err != nil {
+			return nil, 0, err
+		}
+
+		values := make([]int32, 0, len(strValues))
+		for _, value := range strValues {
+			intValue, err := strconv.Atoi(value)
+			if err != nil {
+				return nil, 0, errs.Wrap(err)
+			}
+			values = append(values, int32(intValue))
+		}
+
+		result[userID] = values
+	}
+
+	return result, nextCursor, nil
+}
+
 func (s *userOnline) SetUserOnline(ctx context.Context, userID string, online, offline []int32) error {
 	script := `
 	local key = KEYS[1]

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

@@ -19,8 +19,8 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"github.com/openimsdk/tools/errs"
-	"github.com/openimsdk/tools/utils/stringutil"
 	"github.com/redis/go-redis/v9"
+	"strconv"
 	"time"
 )
 
@@ -58,9 +58,12 @@ func (c *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
 	}
 	mm := make(map[string]int)
 	for k, v := range m {
-		mm[k] = stringutil.StringToInt(v)
+		state, err := strconv.Atoi(v)
+		if err != nil {
+			return nil, errs.WrapMsg(err, "redis token value is not int", "value", v, "userID", userID, "platformID", platformID)
+		}
+		mm[k] = state
 	}
-
 	return mm, nil
 }
 

pkg/common/storage/controller/msg.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/protocol/constant"
 	pbmsg "github.com/openimsdk/protocol/msg"
@@ -47,16 +46,10 @@ const (
 
 // CommonMsgDatabase defines the interface for message database operations.
 type CommonMsgDatabase interface {
-	// BatchInsertChat2DB inserts a batch of messages into the database for a specific conversation.
-	BatchInsertChat2DB(ctx context.Context, conversationID string, msgs []*sdkws.MsgData, currentMaxSeq int64) error
 	// RevokeMsg revokes a message in a conversation.
 	RevokeMsg(ctx context.Context, conversationID string, seq int64, revoke *model.RevokeModel) error
 	// MarkSingleChatMsgsAsRead marks messages as read for a single chat by sequence numbers.
 	MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, seqs []int64) error
-	// DeleteMessagesFromCache deletes message caches from Redis by sequence numbers.
-	DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
-	// BatchInsertChat2Cache increments the sequence number and then batch inserts messages into the cache.
-	BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error)
 	// GetMsgBySeqsRange retrieves messages from MongoDB by a range of sequence numbers.
 	GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
 	// GetMsgBySeqs retrieves messages for large groups from MongoDB by sequence numbers.
@@ -77,7 +70,6 @@ type CommonMsgDatabase interface {
 
 	SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) (err error)
 	SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
-	SetHasReadSeqToDB(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
 	GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error)
 	GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error)
 	UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error
@@ -86,13 +78,11 @@ type CommonMsgDatabase interface {
 	//GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
 	SetSendMsgStatus(ctx context.Context, id string, status int32) error
 	GetSendMsgStatus(ctx context.Context, id string) (int32, error)
-	SearchMessage(ctx context.Context, req *pbmsg.SearchMessageReq) (total int64, msgData []*sdkws.MsgData, err error)
+	SearchMessage(ctx context.Context, req *pbmsg.SearchMessageReq) (total int64, msgData []*pbmsg.SearchedMsgData, err error)
 	FindOneByDocIDs(ctx context.Context, docIDs []string, seqs map[string]int64) (map[string]*sdkws.MsgData, error)
 
 	// to mq
 	MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error
-	MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
-	MsgToMongoMQ(ctx context.Context, key, conversationID string, msgs []*sdkws.MsgData, lastSeq int64) error
 
 	RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*model.UserCount, dateCount map[string]int64, err error)
 	RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*model.GroupCount, dateCount map[string]int64, err error)
@@ -114,22 +104,12 @@ func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser
 	if err != nil {
 		return nil, err
 	}
-	producerToMongo, err := kafka.NewKafkaProducer(conf, kafkaConf.Address, kafkaConf.ToMongoTopic)
-	if err != nil {
-		return nil, err
-	}
-	producerToPush, err := kafka.NewKafkaProducer(conf, kafkaConf.Address, kafkaConf.ToPushTopic)
-	if err != nil {
-		return nil, err
-	}
 	return &commonMsgDatabase{
 		msgDocDatabase:  msgDocModel,
 		msg:             msg,
 		seqUser:         seqUser,
 		seqConversation: seqConversation,
 		producer:        producerToRedis,
-		producerToMongo: producerToMongo,
-		producerToPush:  producerToPush,
 	}, nil
 }
 
@@ -140,8 +120,6 @@ type commonMsgDatabase struct {
 	seqConversation cache.SeqConversationCache
 	seqUser         cache.SeqUser
 	producer        *kafka.Producer
-	producerToMongo *kafka.Producer
-	producerToPush  *kafka.Producer
 }
 
 func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error {
@@ -149,23 +127,6 @@ func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sd
 	return err
 }
 
-func (db *commonMsgDatabase) MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error) {
-	partition, offset, err := db.producerToPush.SendMessage(ctx, key, &pbmsg.PushMsgDataToMQ{MsgData: msg2mq, ConversationID: conversationID})
-	if err != nil {
-		log.ZError(ctx, "MsgToPushMQ", err, "key", key, "msg2mq", msg2mq)
-		return 0, 0, err
-	}
-	return partition, offset, nil
-}
-
-func (db *commonMsgDatabase) MsgToMongoMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData, lastSeq int64) error {
-	if len(messages) > 0 {
-		_, _, err := db.producerToMongo.SendMessage(ctx, key, &pbmsg.MsgDataToMongoByMQ{LastSeq: lastSeq, ConversationID: conversationID, MsgData: messages})
-		return err
-	}
-	return nil
-}
-
 func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationID string, fields []any, key int8, firstSeq int64) error {
 	if len(fields) == 0 {
 		return nil
@@ -267,52 +228,6 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
 	return nil
 }
 
-func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
-	if len(msgList) == 0 {
-		return errs.ErrArgs.WrapMsg("msgList is empty")
-	}
-	msgs := make([]any, len(msgList))
-	for i, msg := range msgList {
-		if msg == nil {
-			continue
-		}
-		var offlinePushModel *model.OfflinePushModel
-		if msg.OfflinePushInfo != nil {
-			offlinePushModel = &model.OfflinePushModel{
-				Title:         msg.OfflinePushInfo.Title,
-				Desc:          msg.OfflinePushInfo.Desc,
-				Ex:            msg.OfflinePushInfo.Ex,
-				IOSPushSound:  msg.OfflinePushInfo.IOSPushSound,
-				IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
-			}
-		}
-		msgs[i] = &model.MsgDataModel{
-			SendID:           msg.SendID,
-			RecvID:           msg.RecvID,
-			GroupID:          msg.GroupID,
-			ClientMsgID:      msg.ClientMsgID,
-			ServerMsgID:      msg.ServerMsgID,
-			SenderPlatformID: msg.SenderPlatformID,
-			SenderNickname:   msg.SenderNickname,
-			SenderFaceURL:    msg.SenderFaceURL,
-			SessionType:      msg.SessionType,
-			MsgFrom:          msg.MsgFrom,
-			ContentType:      msg.ContentType,
-			Content:          string(msg.Content),
-			Seq:              msg.Seq,
-			SendTime:         msg.SendTime,
-			CreateTime:       msg.CreateTime,
-			Status:           msg.Status,
-			Options:          msg.Options,
-			OfflinePush:      offlinePushModel,
-			AtUserIDList:     msg.AtUserIDList,
-			AttachedInfo:     msg.AttachedInfo,
-			Ex:               msg.Ex,
-		}
-	}
-	return db.BatchInsertBlock(ctx, conversationID, msgs, updateKeyMsg, msgList[0].Seq)
-}
-
 func (db *commonMsgDatabase) RevokeMsg(ctx context.Context, conversationID string, seq int64, revoke *model.RevokeModel) error {
 	return db.BatchInsertBlock(ctx, conversationID, []any{revoke}, updateKeyRevoke, seq)
 }
@@ -332,56 +247,6 @@ func (db *commonMsgDatabase) MarkSingleChatMsgsAsRead(ctx context.Context, userI
 	return nil
 }
 
-func (db *commonMsgDatabase) DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error {
-	return db.msg.DeleteMessagesFromCache(ctx, conversationID, seqs)
-}
-
-func (db *commonMsgDatabase) setHasReadSeqs(ctx context.Context, conversationID string, userSeqMap map[string]int64) error {
-	for userID, seq := range userSeqMap {
-		if err := db.seqUser.SetUserReadSeq(ctx, conversationID, userID, seq); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
-	lenList := len(msgs)
-	if int64(lenList) > db.msgTable.GetSingleGocMsgNum() {
-		return 0, false, errs.New("message count exceeds limit", "limit", db.msgTable.GetSingleGocMsgNum()).Wrap()
-	}
-	if lenList < 1 {
-		return 0, false, errs.New("no messages to insert", "minCount", 1).Wrap()
-	}
-	currentMaxSeq, err := db.seqConversation.Malloc(ctx, conversationID, int64(len(msgs)))
-	if err != nil {
-		log.ZError(ctx, "storage.seq.Malloc", err)
-		return 0, false, err
-	}
-	isNew = currentMaxSeq == 0
-	lastMaxSeq := currentMaxSeq
-	userSeqMap := make(map[string]int64)
-	for _, m := range msgs {
-		currentMaxSeq++
-		m.Seq = currentMaxSeq
-		userSeqMap[m.SendID] = m.Seq
-	}
-
-	failedNum, err := db.msg.SetMessagesToCache(ctx, conversationID, msgs)
-	if err != nil {
-		prommetrics.MsgInsertRedisFailedCounter.Add(float64(failedNum))
-		log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
-	} else {
-		prommetrics.MsgInsertRedisSuccessCounter.Inc()
-	}
-	err = db.setHasReadSeqs(ctx, conversationID, userSeqMap)
-	if err != nil {
-		log.ZError(ctx, "SetHasReadSeqs error", err, "userSeqMap", userSeqMap, "conversationID", conversationID)
-		prommetrics.SeqSetFailedCounter.Inc()
-	}
-	return lastMaxSeq, isNew, errs.Wrap(err)
-}
-
 func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
 	for docID, seqs := range db.msgTable.GetDocIDSeqsMap(conversationID, seqs) {
 		// log.ZDebug(ctx, "getMsgBySeqs", "docID", docID, "seqs", seqs)
@@ -581,7 +446,7 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
 
 func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
 	userMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID)
-	if err != nil && errs.Unwrap(err) != redis.Nil {
+	if err != nil {
 		return 0, 0, nil, err
 	}
 	minSeq, err := db.seqConversation.GetMinSeq(ctx, conversationID)
@@ -592,15 +457,28 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	if err != nil {
 		return 0, 0, nil, err
 	}
-	if userMinSeq < minSeq {
+	userMaxSeq, err := db.seqUser.GetUserMaxSeq(ctx, conversationID, userID)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	if userMinSeq > minSeq {
 		minSeq = userMinSeq
 	}
-	var newSeqs []int64
+	if userMaxSeq > 0 && userMaxSeq < maxSeq {
+		maxSeq = userMaxSeq
+	}
+	newSeqs := make([]int64, 0, len(seqs))
 	for _, seq := range seqs {
+		if seq <= 0 {
+			continue
+		}
 		if seq >= minSeq && seq <= maxSeq {
 			newSeqs = append(newSeqs, seq)
 		}
 	}
+	if len(newSeqs) == 0 {
+		return minSeq, maxSeq, nil, nil
+	}
 	successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
 	if err != nil {
 		if err != redis.Nil {
@@ -809,10 +687,6 @@ func (db *commonMsgDatabase) SetHasReadSeq(ctx context.Context, userID string, c
 	return db.seqUser.SetUserReadSeq(ctx, conversationID, userID, hasReadSeq)
 }
 
-func (db *commonMsgDatabase) SetHasReadSeqToDB(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error {
-	return db.seqUser.SetUserReadSeqToDB(ctx, conversationID, userID, hasReadSeq)
-}
-
 func (db *commonMsgDatabase) GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error) {
 	return db.seqUser.GetUserReadSeqs(ctx, userID, conversationIDs)
 }
@@ -886,8 +760,8 @@ func (db *commonMsgDatabase) RangeGroupSendCount(
 	return db.msgDocDatabase.RangeGroupSendCount(ctx, start, end, ase, pageNumber, showNumber)
 }
 
-func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbmsg.SearchMessageReq) (total int64, msgData []*sdkws.MsgData, err error) {
-	var totalMsgs []*sdkws.MsgData
+func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbmsg.SearchMessageReq) (total int64, msgData []*pbmsg.SearchedMsgData, err error) {
+	var totalMsgs []*pbmsg.SearchedMsgData
 	total, msgs, err := db.msgDocDatabase.SearchMessage(ctx, req)
 	if err != nil {
 		return 0, nil, err
@@ -896,7 +770,13 @@ func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbmsg.Searc
 		if msg.IsRead {
 			msg.Msg.IsRead = true
 		}
-		totalMsgs = append(totalMsgs, convert.MsgDB2Pb(msg.Msg))
+		searchedMsgData := &pbmsg.SearchedMsgData{MsgData: convert.MsgDB2Pb(msg.Msg)}
+
+		if msg.Revoke != nil {
+			searchedMsgData.IsRevoked = true
+		}
+
+		totalMsgs = append(totalMsgs, searchedMsgData)
 	}
 	return total, totalMsgs, nil
 }

pkg/common/storage/controller/msg_transfer.go

@@ -0,0 +1,286 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	pbmsg "github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mq/kafka"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type MsgTransferDatabase interface {
+	// BatchInsertChat2DB inserts a batch of messages into the database for a specific conversation.
+	BatchInsertChat2DB(ctx context.Context, conversationID string, msgs []*sdkws.MsgData, currentMaxSeq int64) error
+	// DeleteMessagesFromCache deletes message caches from Redis by sequence numbers.
+	DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
+
+	// BatchInsertChat2Cache increments the sequence number and then batch inserts messages into the cache.
+	BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error)
+	SetHasReadSeqToDB(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
+
+	// to mq
+	MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
+	MsgToMongoMQ(ctx context.Context, key, conversationID string, msgs []*sdkws.MsgData, lastSeq int64) error
+}
+
+func NewMsgTransferDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, kafkaConf *config.Kafka) (MsgTransferDatabase, error) {
+	conf, err := kafka.BuildProducerConfig(*kafkaConf.Build())
+	if err != nil {
+		return nil, err
+	}
+	producerToMongo, err := kafka.NewKafkaProducer(conf, kafkaConf.Address, kafkaConf.ToMongoTopic)
+	if err != nil {
+		return nil, err
+	}
+	producerToPush, err := kafka.NewKafkaProducer(conf, kafkaConf.Address, kafkaConf.ToPushTopic)
+	if err != nil {
+		return nil, err
+	}
+	return &msgTransferDatabase{
+		msgDocDatabase:  msgDocModel,
+		msg:             msg,
+		seqUser:         seqUser,
+		seqConversation: seqConversation,
+		producerToMongo: producerToMongo,
+		producerToPush:  producerToPush,
+	}, nil
+}
+
+type msgTransferDatabase struct {
+	msgDocDatabase  database.Msg
+	msgTable        model.MsgDocModel
+	msg             cache.MsgCache
+	seqConversation cache.SeqConversationCache
+	seqUser         cache.SeqUser
+	producerToMongo *kafka.Producer
+	producerToPush  *kafka.Producer
+}
+
+func (db *msgTransferDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
+	if len(msgList) == 0 {
+		return errs.ErrArgs.WrapMsg("msgList is empty")
+	}
+	msgs := make([]any, len(msgList))
+	for i, msg := range msgList {
+		if msg == nil {
+			continue
+		}
+		var offlinePushModel *model.OfflinePushModel
+		if msg.OfflinePushInfo != nil {
+			offlinePushModel = &model.OfflinePushModel{
+				Title:         msg.OfflinePushInfo.Title,
+				Desc:          msg.OfflinePushInfo.Desc,
+				Ex:            msg.OfflinePushInfo.Ex,
+				IOSPushSound:  msg.OfflinePushInfo.IOSPushSound,
+				IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
+			}
+		}
+		msgs[i] = &model.MsgDataModel{
+			SendID:           msg.SendID,
+			RecvID:           msg.RecvID,
+			GroupID:          msg.GroupID,
+			ClientMsgID:      msg.ClientMsgID,
+			ServerMsgID:      msg.ServerMsgID,
+			SenderPlatformID: msg.SenderPlatformID,
+			SenderNickname:   msg.SenderNickname,
+			SenderFaceURL:    msg.SenderFaceURL,
+			SessionType:      msg.SessionType,
+			MsgFrom:          msg.MsgFrom,
+			ContentType:      msg.ContentType,
+			Content:          string(msg.Content),
+			Seq:              msg.Seq,
+			SendTime:         msg.SendTime,
+			CreateTime:       msg.CreateTime,
+			Status:           msg.Status,
+			Options:          msg.Options,
+			OfflinePush:      offlinePushModel,
+			AtUserIDList:     msg.AtUserIDList,
+			AttachedInfo:     msg.AttachedInfo,
+			Ex:               msg.Ex,
+		}
+	}
+	return db.BatchInsertBlock(ctx, conversationID, msgs, updateKeyMsg, msgList[0].Seq)
+}
+
+func (db *msgTransferDatabase) BatchInsertBlock(ctx context.Context, conversationID string, fields []any, key int8, firstSeq int64) error {
+	if len(fields) == 0 {
+		return nil
+	}
+	num := db.msgTable.GetSingleGocMsgNum()
+	// num = 100
+	for i, field := range fields { // Check the type of the field
+		var ok bool
+		switch key {
+		case updateKeyMsg:
+			var msg *model.MsgDataModel
+			msg, ok = field.(*model.MsgDataModel)
+			if msg != nil && msg.Seq != firstSeq+int64(i) {
+				return errs.ErrInternalServer.WrapMsg("seq is invalid")
+			}
+		case updateKeyRevoke:
+			_, ok = field.(*model.RevokeModel)
+		default:
+			return errs.ErrInternalServer.WrapMsg("key is invalid")
+		}
+		if !ok {
+			return errs.ErrInternalServer.WrapMsg("field type is invalid")
+		}
+	}
+	// Returns true if the document exists in the database, false if the document does not exist in the database
+	updateMsgModel := func(seq int64, i int) (bool, error) {
+		var (
+			res *mongo.UpdateResult
+			err error
+		)
+		docID := db.msgTable.GetDocID(conversationID, seq)
+		index := db.msgTable.GetMsgIndex(seq)
+		field := fields[i]
+		switch key {
+		case updateKeyMsg:
+			res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "msg", field)
+		case updateKeyRevoke:
+			res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "revoke", field)
+		}
+		if err != nil {
+			return false, err
+		}
+		return res.MatchedCount > 0, nil
+	}
+	tryUpdate := true
+	for i := 0; i < len(fields); i++ {
+		seq := firstSeq + int64(i) // Current sequence number
+		if tryUpdate {
+			matched, err := updateMsgModel(seq, i)
+			if err != nil {
+				return err
+			}
+			if matched {
+				continue // The current data has been updated, skip the current data
+			}
+		}
+		doc := model.MsgDocModel{
+			DocID: db.msgTable.GetDocID(conversationID, seq),
+			Msg:   make([]*model.MsgInfoModel, num),
+		}
+		var insert int // Inserted data number
+		for j := i; j < len(fields); j++ {
+			seq = firstSeq + int64(j)
+			if db.msgTable.GetDocID(conversationID, seq) != doc.DocID {
+				break
+			}
+			insert++
+			switch key {
+			case updateKeyMsg:
+				doc.Msg[db.msgTable.GetMsgIndex(seq)] = &model.MsgInfoModel{
+					Msg: fields[j].(*model.MsgDataModel),
+				}
+			case updateKeyRevoke:
+				doc.Msg[db.msgTable.GetMsgIndex(seq)] = &model.MsgInfoModel{
+					Revoke: fields[j].(*model.RevokeModel),
+				}
+			}
+		}
+		for i, msgInfo := range doc.Msg {
+			if msgInfo == nil {
+				msgInfo = &model.MsgInfoModel{}
+				doc.Msg[i] = msgInfo
+			}
+			if msgInfo.DelList == nil {
+				doc.Msg[i].DelList = []string{}
+			}
+		}
+		if err := db.msgDocDatabase.Create(ctx, &doc); err != nil {
+			if mongo.IsDuplicateKeyError(err) {
+				i--              // already inserted
+				tryUpdate = true // next block use update mode
+				continue
+			}
+			return err
+		}
+		tryUpdate = false // The current block is inserted successfully, and the next block is inserted preferentially
+		i += insert - 1   // Skip the inserted data
+	}
+	return nil
+}
+
+func (db *msgTransferDatabase) DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error {
+	return db.msg.DeleteMessagesFromCache(ctx, conversationID, seqs)
+}
+
+func (db *msgTransferDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
+	lenList := len(msgs)
+	if int64(lenList) > db.msgTable.GetSingleGocMsgNum() {
+		return 0, false, errs.New("message count exceeds limit", "limit", db.msgTable.GetSingleGocMsgNum()).Wrap()
+	}
+	if lenList < 1 {
+		return 0, false, errs.New("no messages to insert", "minCount", 1).Wrap()
+	}
+	currentMaxSeq, err := db.seqConversation.Malloc(ctx, conversationID, int64(len(msgs)))
+	if err != nil {
+		log.ZError(ctx, "storage.seq.Malloc", err)
+		return 0, false, err
+	}
+	isNew = currentMaxSeq == 0
+	lastMaxSeq := currentMaxSeq
+	userSeqMap := make(map[string]int64)
+	for _, m := range msgs {
+		currentMaxSeq++
+		m.Seq = currentMaxSeq
+		userSeqMap[m.SendID] = m.Seq
+	}
+
+	failedNum, err := db.msg.SetMessagesToCache(ctx, conversationID, msgs)
+	if err != nil {
+		prommetrics.MsgInsertRedisFailedCounter.Add(float64(failedNum))
+		log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
+	} else {
+		prommetrics.MsgInsertRedisSuccessCounter.Inc()
+	}
+	err = db.setHasReadSeqs(ctx, conversationID, userSeqMap)
+	if err != nil {
+		log.ZError(ctx, "SetHasReadSeqs error", err, "userSeqMap", userSeqMap, "conversationID", conversationID)
+		prommetrics.SeqSetFailedCounter.Inc()
+	}
+	return lastMaxSeq, isNew, errs.Wrap(err)
+}
+
+func (db *msgTransferDatabase) setHasReadSeqs(ctx context.Context, conversationID string, userSeqMap map[string]int64) error {
+	for userID, seq := range userSeqMap {
+		if err := db.seqUser.SetUserReadSeq(ctx, conversationID, userID, seq); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (db *msgTransferDatabase) SetHasReadSeqToDB(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error {
+	return db.seqUser.SetUserReadSeqToDB(ctx, conversationID, userID, hasReadSeq)
+}
+
+func (db *msgTransferDatabase) MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error) {
+	partition, offset, err := db.producerToPush.SendMessage(ctx, key, &pbmsg.PushMsgDataToMQ{MsgData: msg2mq, ConversationID: conversationID})
+	if err != nil {
+		log.ZError(ctx, "MsgToPushMQ", err, "key", key, "msg2mq", msg2mq)
+		return 0, 0, err
+	}
+	return partition, offset, nil
+}
+
+func (db *msgTransferDatabase) MsgToMongoMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData, lastSeq int64) error {
+	if len(messages) > 0 {
+		_, _, err := db.producerToMongo.SendMessage(ctx, key, &pbmsg.MsgDataToMongoByMQ{LastSeq: lastSeq, ConversationID: conversationID, MsgData: messages})
+		if err != nil {
+			log.ZError(ctx, "MsgToMongoMQ", err, "key", key, "conversationID", conversationID, "lastSeq", lastSeq)
+			return err
+		}
+	}
+	return nil
+}

pkg/common/storage/controller/push.go

@@ -17,21 +17,45 @@ package controller
 import (
 	"context"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/protocol/push"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mq/kafka"
 )
 
 type PushDatabase interface {
 	DelFcmToken(ctx context.Context, userID string, platformID int) error
+	MsgToOfflinePushMQ(ctx context.Context, key string, userIDs []string, msg2mq *sdkws.MsgData) error
 }
 
 type pushDataBase struct {
-	cache cache.ThirdCache
+	cache                 cache.ThirdCache
+	producerToOfflinePush *kafka.Producer
 }
 
-func NewPushDatabase(cache cache.ThirdCache) PushDatabase {
-	return &pushDataBase{cache: cache}
+func NewPushDatabase(cache cache.ThirdCache, kafkaConf *config.Kafka) PushDatabase {
+	conf, err := kafka.BuildProducerConfig(*kafkaConf.Build())
+	if err != nil {
+		return nil
+	}
+	producerToOfflinePush, err := kafka.NewKafkaProducer(conf, kafkaConf.Address, kafkaConf.ToOfflinePushTopic)
+	if err != nil {
+		return nil
+	}
+	return &pushDataBase{
+		cache:                 cache,
+		producerToOfflinePush: producerToOfflinePush,
+	}
 }
 
 func (p *pushDataBase) DelFcmToken(ctx context.Context, userID string, platformID int) error {
 	return p.cache.DelFcmToken(ctx, userID, platformID)
 }
+
+func (p *pushDataBase) MsgToOfflinePushMQ(ctx context.Context, key string, userIDs []string, msg2mq *sdkws.MsgData) error {
+	_, _, err := p.producerToOfflinePush.SendMessage(ctx, key, &push.PushMsgReq{MsgData: msg2mq, UserIDs: userIDs})
+	log.ZInfo(ctx, "message is push to offlinePush topic", "key", key, "userIDs", userIDs, "msg", msg2mq.String())
+	return err
+}

pkg/common/storage/controller/user.go

@@ -16,12 +16,13 @@ package controller
 
 import (
 	"context"
+	"time"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/tools/db/pagination"
 	"github.com/openimsdk/tools/db/tx"
 	"github.com/openimsdk/tools/utils/datautil"
-	"time"
 
 	"github.com/openimsdk/protocol/user"
 	"github.com/openimsdk/tools/errs"
@@ -111,10 +112,14 @@ func (u *userDatabase) InitOnce(ctx context.Context, users []*model.User) error
 // FindWithError Get the information of the specified user and return an error if the userID is not found.
 func (u *userDatabase) FindWithError(ctx context.Context, userIDs []string) (users []*model.User, err error) {
 	userIDs = datautil.Distinct(userIDs)
+
+	// TODO: Add logic to identify which user IDs are distinct and which user IDs were not found.
+
 	users, err = u.cache.GetUsersInfo(ctx, userIDs)
 	if err != nil {
 		return
 	}
+
 	if len(users) != len(userIDs) {
 		err = errs.ErrRecordNotFound.WrapMsg("userID not found")
 	}

pkg/localcache/lru/lru.go

@@ -20,7 +20,9 @@ type EvictCallback[K comparable, V any] simplelru.EvictCallback[K, V]
 
 type LRU[K comparable, V any] interface {
 	Get(key K, fetch func() (V, error)) (V, error)
+	Set(key K, value V)
 	SetHas(key K, value V) bool
+	GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error)
 	Del(key K) bool
 	Stop()
 }

pkg/localcache/lru/lru_expiration.go

@@ -51,6 +51,11 @@ type ExpirationLRU[K comparable, V any] struct {
 	target     Target
 }
 
+func (x *ExpirationLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 	x.lock.Lock()
 	v, ok := x.core.Get(key)
@@ -99,5 +104,11 @@ func (x *ExpirationLRU[K, V]) SetHas(key K, value V) bool {
 	return false
 }
 
+func (x *ExpirationLRU[K, V]) Set(key K, value V) {
+	x.lock.Lock()
+	defer x.lock.Unlock()
+	x.core.Add(key, &expirationLruItem[V]{value: value})
+}
+
 func (x *ExpirationLRU[K, V]) Stop() {
 }

pkg/localcache/lru/lru_lazy.go

@@ -88,18 +88,75 @@ func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 	return v.value, v.err
 }
 
-//func (x *LayLRU[K, V]) Set(key K, value V) {
-//	x.lock.Lock()
-//	x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
-//	x.lock.Unlock()
-//}
-//
+func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
+	var (
+		err  error
+		once sync.Once
+	)
+
+	res := make(map[K]V)
+	queries := make([]K, 0)
+	setVs := make(map[K]*layLruItem[V])
+	for _, key := range keys {
+		x.lock.Lock()
+		v, ok := x.core.Get(key)
+		x.lock.Unlock()
+		if ok {
+			v.lock.Lock()
+			expires, value, err1 := v.expires, v.value, v.err
+			v.lock.Unlock()
+			if expires != 0 && expires > time.Now().UnixMilli() {
+				x.target.IncrGetHit()
+				res[key] = value
+				if err1 != nil {
+					once.Do(func() {
+						err = err1
+					})
+				}
+				continue
+			}
+		}
+		queries = append(queries, key)
+	}
+	values, err1 := fetch(queries)
+	if err1 != nil {
+		once.Do(func() {
+			err = err1
+		})
+	}
+	for key, val := range values {
+		v := &layLruItem[V]{}
+		v.value = val
+
+		if err == nil {
+			v.expires = time.Now().Add(x.successTTL).UnixMilli()
+			x.target.IncrGetSuccess()
+		} else {
+			v.expires = time.Now().Add(x.failedTTL).UnixMilli()
+			x.target.IncrGetFailed()
+		}
+		setVs[key] = v
+		x.lock.Lock()
+		x.core.Add(key, v)
+		x.lock.Unlock()
+		res[key] = val
+	}
+
+	return res, err
+}
+
 //func (x *LayLRU[K, V]) Has(key K) bool {
 //	x.lock.Lock()
 //	defer x.lock.Unlock()
 //	return x.core.Contains(key)
 //}
 
+func (x *LayLRU[K, V]) Set(key K, value V) {
+	x.lock.Lock()
+	defer x.lock.Unlock()
+	x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
+}
+
 func (x *LayLRU[K, V]) SetHas(key K, value V) bool {
 	x.lock.Lock()
 	defer x.lock.Unlock()

pkg/localcache/lru/lru_slot.go

@@ -32,6 +32,29 @@ type slotLRU[K comparable, V any] struct {
 	hash  func(k K) uint64
 }
 
+func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
+	var (
+		slotKeys = make(map[uint64][]K)
+		vs       = make(map[K]V)
+	)
+
+	for _, k := range keys {
+		index := x.getIndex(k)
+		slotKeys[index] = append(slotKeys[index], k)
+	}
+
+	for k, v := range slotKeys {
+		batches, err := x.slots[k].GetBatch(v, fetch)
+		if err != nil {
+			return nil, err
+		}
+		for key, value := range batches {
+			vs[key] = value
+		}
+	}
+	return vs, nil
+}
+
 func (x *slotLRU[K, V]) getIndex(k K) uint64 {
 	return x.hash(k) % x.n
 }
@@ -40,6 +63,10 @@ func (x *slotLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 	return x.slots[x.getIndex(key)].Get(key, fetch)
 }
 
+func (x *slotLRU[K, V]) Set(key K, value V) {
+	x.slots[x.getIndex(key)].Set(key, value)
+}
+
 func (x *slotLRU[K, V]) SetHas(key K, value V) bool {
 	return x.slots[x.getIndex(key)].SetHas(key, value)
 }

pkg/rpccache/conversation.go

@@ -86,7 +86,7 @@ func (c *ConversationLocalCache) GetConversation(ctx context.Context, userID, co
 		if err == nil {
 			log.ZDebug(ctx, "ConversationLocalCache GetConversation return", "userID", userID, "conversationID", conversationID, "value", val)
 		} else {
-			log.ZError(ctx, "ConversationLocalCache GetConversation return", err, "userID", userID, "conversationID", conversationID)
+			log.ZWarn(ctx, "ConversationLocalCache GetConversation return", err, "userID", userID, "conversationID", conversationID)
 		}
 	}()
 	var cache cacheProto[pbconversation.Conversation]

pkg/rpccache/online.go

@@ -2,60 +2,197 @@ package rpccache
 
 import (
 	"context"
+	"fmt"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/protocol/user"
+	"math/rand"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/open-im-server/v3/pkg/util/useronline"
+	"github.com/openimsdk/tools/db/cacheutil"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/redis/go-redis/v9"
-	"math/rand"
-	"strconv"
-	"time"
 )
 
-func NewOnlineCache(user rpcclient.UserRpcClient, group *GroupLocalCache, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) *OnlineCache {
+func NewOnlineCache(user rpcclient.UserRpcClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) {
+	l := &sync.Mutex{}
 	x := &OnlineCache{
-		user:  user,
-		group: group,
-		local: lru.NewSlotLRU(1024, localcache.LRUStringHash, func() lru.LRU[string, []int32] {
-			return lru.NewLayLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
-		}),
+		user:          user,
+		group:         group,
+		fullUserCache: fullUserCache,
+		Lock:          l,
+		Cond:          sync.NewCond(l),
 	}
+
+	ctx := mcontext.SetOperationID(context.TODO(), strconv.FormatInt(time.Now().UnixNano()+int64(rand.Uint32()), 10))
+
+	switch x.fullUserCache {
+	case true:
+		log.ZDebug(ctx, "fullUserCache is true")
+		x.mapCache = cacheutil.NewCache[string, []int32]()
+		go func() {
+			if err := x.initUsersOnlineStatus(ctx); err != nil {
+				log.ZError(ctx, "initUsersOnlineStatus failed", err)
+			}
+		}()
+	case false:
+		log.ZDebug(ctx, "fullUserCache is false")
+		x.lruCache = lru.NewSlotLRU(1024, localcache.LRUStringHash, func() lru.LRU[string, []int32] {
+			return lru.NewLayLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
+		})
+		x.CurrentPhase.Store(DoSubscribeOver)
+		x.Cond.Broadcast()
+	}
+
 	go func() {
-		ctx := mcontext.SetOperationID(context.Background(), cachekey.OnlineChannel+strconv.FormatUint(rand.Uint64(), 10))
-		for message := range rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel() {
-			userID, platformIDs, err := useronline.ParseUserOnlineStatus(message.Payload)
-			if err != nil {
-				log.ZError(ctx, "OnlineCache setUserOnline redis subscribe parseUserOnlineStatus", err, "payload", message.Payload, "channel", message.Channel)
-				continue
-			}
-			storageCache := x.setUserOnline(userID, platformIDs)
-			log.ZDebug(ctx, "OnlineCache setUserOnline", "userID", userID, "platformIDs", platformIDs, "payload", message.Payload, "storageCache", storageCache)
-			if fn != nil {
-				fn(ctx, userID, platformIDs)
-			}
-		}
+		x.doSubscribe(ctx, rdb, fn)
 	}()
-	return x
+	return x, nil
 }
 
+const (
+	Begin uint32 = iota
+	DoOnlineStatusOver
+	DoSubscribeOver
+)
+
 type OnlineCache struct {
 	user  rpcclient.UserRpcClient
 	group *GroupLocalCache
-	local lru.LRU[string, []int32]
+
+	// fullUserCache if enabled, caches the online status of all users using mapCache;
+	// otherwise, only a portion of users' online statuses (regardless of whether they are online) will be cached using lruCache.
+	fullUserCache bool
+
+	lruCache lru.LRU[string, []int32]
+	mapCache *cacheutil.Cache[string, []int32]
+
+	Lock         *sync.Mutex
+	Cond         *sync.Cond
+	CurrentPhase atomic.Uint32
+}
+
+func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
+	log.ZDebug(ctx, "init users online status begin")
+
+	var (
+		totalSet      atomic.Int64
+		maxTries      = 5
+		retryInterval = time.Second * 5
+
+		resp *user.GetAllOnlineUsersResp
+	)
+
+	defer func(t time.Time) {
+		log.ZInfo(ctx, "init users online status end", "cost", time.Since(t), "totalSet", totalSet.Load())
+		o.CurrentPhase.Store(DoOnlineStatusOver)
+		o.Cond.Broadcast()
+	}(time.Now())
+
+	retryOperation := func(operation func() error, operationName string) error {
+		for i := 0; i < maxTries; i++ {
+			if err = operation(); err != nil {
+				log.ZWarn(ctx, fmt.Sprintf("initUsersOnlineStatus: %s failed", operationName), err)
+				time.Sleep(retryInterval)
+			} else {
+				return nil
+			}
+		}
+		return err
+	}
+
+	cursor := uint64(0)
+	for resp == nil || resp.NextCursor != 0 {
+		if err = retryOperation(func() error {
+			resp, err = o.user.GetAllOnlineUsers(ctx, cursor)
+			if err != nil {
+				return err
+			}
+
+			for _, u := range resp.StatusList {
+				if u.Status == constant.Online {
+					o.setUserOnline(u.UserID, u.PlatformIDs)
+				}
+				totalSet.Add(1)
+			}
+			cursor = resp.NextCursor
+			return nil
+		}, "getAllOnlineUsers"); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
+	o.Lock.Lock()
+	ch := rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel()
+	for o.CurrentPhase.Load() < DoOnlineStatusOver {
+		o.Cond.Wait()
+	}
+	o.Lock.Unlock()
+	log.ZInfo(ctx, "begin doSubscribe")
+
+	doMessage := func(message *redis.Message) {
+		userID, platformIDs, err := useronline.ParseUserOnlineStatus(message.Payload)
+		if err != nil {
+			log.ZError(ctx, "OnlineCache setHasUserOnline redis subscribe parseUserOnlineStatus", err, "payload", message.Payload, "channel", message.Channel)
+			return
+		}
+		log.ZDebug(ctx, fmt.Sprintf("get subscribe %s message", cachekey.OnlineChannel), "useID", userID, "platformIDs", platformIDs)
+		switch o.fullUserCache {
+		case true:
+			if len(platformIDs) == 0 {
+				// offline
+				o.mapCache.Delete(userID)
+			} else {
+				o.mapCache.Store(userID, platformIDs)
+			}
+		case false:
+			storageCache := o.setHasUserOnline(userID, platformIDs)
+			log.ZDebug(ctx, "OnlineCache setHasUserOnline", "userID", userID, "platformIDs", platformIDs, "payload", message.Payload, "storageCache", storageCache)
+			if fn != nil {
+				fn(ctx, userID, platformIDs)
+			}
+		}
+	}
+
+	if o.CurrentPhase.Load() == DoOnlineStatusOver {
+		for done := false; !done; {
+			select {
+			case message := <-ch:
+				doMessage(message)
+			default:
+				o.CurrentPhase.Store(DoSubscribeOver)
+				o.Cond.Broadcast()
+				done = true
+			}
+		}
+	}
+
+	for message := range ch {
+		doMessage(message)
+	}
 }
 
 func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
-	platformIDs, err := o.local.Get(userID, func() ([]int32, error) {
+	platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) {
 		return o.user.GetUserOnlinePlatform(ctx, userID)
 	})
 	if err != nil {
 		log.ZError(ctx, "OnlineCache GetUserOnlinePlatform", err, "userID", userID)
 		return nil, err
 	}
-	log.ZDebug(ctx, "OnlineCache GetUserOnlinePlatform", "userID", userID, "platformIDs", platformIDs)
+	//log.ZDebug(ctx, "OnlineCache GetUserOnlinePlatform", "userID", userID, "platformIDs", platformIDs)
 	return platformIDs, nil
 }
 
@@ -69,6 +206,16 @@ func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string)
 	return platformIDs, nil
 }
 
+// func (o *OnlineCache) GetUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string]int32, error) {
+// 	platformIDs, err := o.getUserOnlinePlatform(ctx, userIDs)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+// 	tmp := make([]int32, len(platformIDs))
+// 	copy(tmp, platformIDs)
+// 	return platformIDs, nil
+// }
+
 func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
 	platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
 	if err != nil {
@@ -77,10 +224,68 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
 	return len(platformIDs) > 0, nil
 }
 
+func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
+	platformIDsMap, err := o.lruCache.GetBatch(userIDs, func(missingUsers []string) (map[string][]int32, error) {
+		platformIDsMap := make(map[string][]int32)
+
+		usersStatus, err := o.user.GetUsersOnlinePlatform(ctx, missingUsers)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, u := range usersStatus {
+			platformIDsMap[u.UserID] = u.PlatformIDs
+		}
+
+		return platformIDsMap, nil
+	})
+	if err != nil {
+		log.ZError(ctx, "OnlineCache GetUserOnlinePlatform", err, "userID", userIDs)
+		return nil, err
+	}
+	return platformIDsMap, nil
+}
+
+func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
+	t := time.Now()
+
+	var (
+		onlineUserIDs  = make([]string, 0, len(userIDs))
+		offlineUserIDs = make([]string, 0, len(userIDs))
+	)
+
+	switch o.fullUserCache {
+	case true:
+		for _, userID := range userIDs {
+			if _, ok := o.mapCache.Load(userID); ok {
+				onlineUserIDs = append(onlineUserIDs, userID)
+			} else {
+				offlineUserIDs = append(offlineUserIDs, userID)
+			}
+		}
+	case false:
+		userOnlineMap, err := o.getUserOnlinePlatformBatch(ctx, userIDs)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		for key, value := range userOnlineMap {
+			if len(value) > 0 {
+				onlineUserIDs = append(onlineUserIDs, key)
+			} else {
+				offlineUserIDs = append(offlineUserIDs, key)
+			}
+		}
+	}
+
+	log.ZInfo(ctx, "get users online", "online users length", len(userIDs), "offline users length", len(offlineUserIDs), "cost", time.Since(t))
+	return userIDs, offlineUserIDs, nil
+}
+
 //func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, error) {
 //	onlineUserIDs := make([]string, 0, len(userIDs))
 //	for _, userID := range userIDs {
-//		online, err := o.GetUserOnline(ctx, userID)
+// online, err := o.GetUserOnline(ctx, userID)
 //		if err != nil {
 //			return nil, err
 //		}
@@ -111,6 +316,15 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
 //	return onlineUserIDs, nil
 //}
 
-func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) bool {
-	return o.local.SetHas(userID, platformIDs)
+func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
+	switch o.fullUserCache {
+	case true:
+		o.mapCache.Store(userID, platformIDs)
+	case false:
+		o.lruCache.Set(userID, platformIDs)
+	}
+}
+
+func (o *OnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
+	return o.lruCache.SetHas(userID, platformIDs)
 }

pkg/rpcclient/msg.go

@@ -17,6 +17,11 @@ package rpcclient
 import (
 	"context"
 	"encoding/json"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/msg"
@@ -28,9 +33,6 @@ import (
 	"github.com/openimsdk/tools/utils/idutil"
 	"github.com/openimsdk/tools/utils/jsonutil"
 	"github.com/openimsdk/tools/utils/timeutil"
-	"google.golang.org/grpc"
-	"google.golang.org/protobuf/proto"
-	"time"
 )
 
 func newContentTypeConf(conf *config.Notification) map[int32]config.NotificationConfig {
@@ -221,6 +223,19 @@ func (m *MessageRpcClient) PullMessageBySeqList(ctx context.Context, req *sdkws.
 	return resp, nil
 }
 
+func (m *MessageRpcClient) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *msg.GetConversationsHasReadAndMaxSeqReq) (*msg.GetConversationsHasReadAndMaxSeqResp, error) {
+	resp, err := m.Client.GetConversationsHasReadAndMaxSeq(ctx, req)
+	if err != nil {
+		// Wrap the error to provide more context if the gRPC call fails.
+		return nil, err
+	}
+	return resp, nil
+}
+
+func (m *MessageRpcClient) GetSeqMessage(ctx context.Context, req *msg.GetSeqMessageReq) (*msg.GetSeqMessageResp, error) {
+	return m.Client.GetSeqMessage(ctx, req)
+}
+
 func (m *MessageRpcClient) GetConversationMaxSeq(ctx context.Context, conversationID string) (int64, error) {
 	resp, err := m.Client.GetConversationMaxSeq(ctx, &msg.GetConversationMaxSeqReq{ConversationID: conversationID})
 	if err != nil {

pkg/rpcclient/user.go

@@ -169,6 +169,15 @@ func (u *UserRpcClient) Access(ctx context.Context, ownerUserID string) error {
 	return authverify.CheckAccessV3(ctx, ownerUserID, u.imAdminUserID)
 }
 
+// GetAllUserID retrieves all user IDs with pagination options.
+func (u *UserRpcClient) GetAllUserID(ctx context.Context, pageNumber, showNumber int32) (*user.GetAllUserIDResp, error) {
+	resp, err := u.Client.GetAllUserID(ctx, &user.GetAllUserIDReq{Pagination: &sdkws.RequestPagination{PageNumber: pageNumber, ShowNumber: showNumber}})
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
 // GetAllUserIDs retrieves all user IDs with pagination options.
 func (u *UserRpcClient) GetAllUserIDs(ctx context.Context, pageNumber, showNumber int32) ([]string, error) {
 	resp, err := u.Client.GetAllUserID(ctx, &user.GetAllUserIDReq{Pagination: &sdkws.RequestPagination{PageNumber: pageNumber, ShowNumber: showNumber}})
@@ -215,3 +224,7 @@ func (u *UserRpcClient) GetUserOnlinePlatform(ctx context.Context, userID string
 	}
 	return resp[0].PlatformIDs, nil
 }
+
+func (u *UserRpcClient) GetAllOnlineUsers(ctx context.Context, cursor uint64) (*user.GetAllOnlineUsersResp, error) {
+	return u.Client.GetAllOnlineUsers(ctx, &user.GetAllOnlineUsersReq{Cursor: cursor})
+}

pkg/util/conversationutil/conversationutil.go

@@ -19,6 +19,14 @@ func GenGroupConversationID(groupID string) string {
 	return "sg_" + groupID
 }
 
+func IsGroupConversationID(conversationID string) bool {
+	return strings.HasPrefix(conversationID, "sg_")
+}
+
+func IsNotificationConversationID(conversationID string) bool {
+	return strings.HasPrefix(conversationID, "n_")
+}
+
 func GenConversationUniqueKeyForSingle(sendID, recvID string) string {
 	l := []string{sendID, recvID}
 	sort.Strings(l)