From 1f3fa72fe3d0bbc49c757bbb2162b40d485521d1 Mon Sep 17 00:00:00 2001 From: icey-yu <1186114839@qq.com> Date: Thu, 22 Aug 2024 16:52:31 +0800 Subject: [PATCH 1/4] feat: add EnableHistoryForNewMembers --- config/openim-rpc-group.yml | 3 ++ go.mod | 5 +-- go.sum | 6 +-- internal/rpc/group/group.go | 34 ++++++++++++----- internal/rpc/group/notification.go | 60 ++++++++++++++++-------------- pkg/common/config/config.go | 3 +- 6 files changed, 65 insertions(+), 46 deletions(-) diff --git a/config/openim-rpc-group.yml b/config/openim-rpc-group.yml index f49c46207f..5fff8b9fe3 100644 --- a/config/openim-rpc-group.yml +++ b/config/openim-rpc-group.yml @@ -11,3 +11,6 @@ prometheus: enable: true # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup ports: [ 20103 ] + + +enableHistoryForNewMembers: true \ No newline at end of file diff --git a/go.mod b/go.mod index fba1499fe1..6d501e0525 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/gorilla/websocket v1.5.1 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/mitchellh/mapstructure v1.5.0 - github.com/openimsdk/protocol v0.0.69 + github.com/openimsdk/protocol v0.0.72-alpha.1 github.com/openimsdk/tools v0.0.49-alpha.55 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 @@ -41,6 +41,7 @@ require ( github.com/spf13/viper v1.18.2 github.com/stathat/consistent v1.0.0 go.uber.org/automaxprocs v1.5.3 + golang.org/x/exp v0.0.0-20230905200255-921286631fa9 golang.org/x/sync v0.6.0 ) @@ -74,7 +75,6 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/bytedance/sonic v1.9.1 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect - github.com/chai2010/webp v1.1.1 // indirect github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect github.com/clbanning/mxj v1.8.4 // indirect github.com/coreos/go-semver v0.3.0 // indirect @@ -170,7 +170,6 @@ require ( go.uber.org/atomic v1.9.0 // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/arch v0.3.0 // indirect - golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect golang.org/x/image v0.15.0 // indirect golang.org/x/net v0.22.0 // indirect golang.org/x/oauth2 v0.17.0 // indirect diff --git a/go.sum b/go.sum index 1a8e1d76d8..1c011e42e5 100644 --- a/go.sum +++ b/go.sum @@ -71,8 +71,6 @@ github.com/bytedance/sonic v1.9.1/go.mod h1:i736AoUSYt75HyZLoJW9ERYxcy6eaN6h4BZX github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/chai2010/webp v1.1.1 h1:jTRmEccAJ4MGrhFOrPMpNGIJ/eybIgwKpcACsrTEapk= -github.com/chai2010/webp v1.1.1/go.mod h1:0XVwvZWdjjdxpUEIf7b9g9VkHFnInUSYujwqTLEuldU= github.com/chenzhuoyu/base64x v0.0.0-20211019084208-fb5309c8db06/go.mod h1:DH46F32mSOjUmXrMHnKwZdA8wcEefY7UVqBKYGjpdQY= github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 h1:qSGYFH7+jGhDF8vLC+iwCD4WpbV1EBDSzWkJODFLams= github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311/go.mod h1:b583jCggY9gE99b6G5LEC39OIiVsWj+R97kbl5odCEk= @@ -321,8 +319,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y= github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM= github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y= github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= -github.com/openimsdk/protocol v0.0.69 h1:dVi8meSg8kmUzSH1XQab4MjihqKkkcCAmt1BYXPJuXo= -github.com/openimsdk/protocol v0.0.69/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= +github.com/openimsdk/protocol v0.0.72-alpha.1 h1:qaNV6+mBme2NC9lXWAScgKj5gjAcihKixKzxTndDzvw= +github.com/openimsdk/protocol v0.0.72-alpha.1/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= github.com/openimsdk/tools v0.0.49-alpha.55 h1:KPgC53oqiwZYssLKljhtXbWXifMlTj2SSQEusj4Uf4k= github.com/openimsdk/tools v0.0.49-alpha.55/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4= github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4= diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index aa12c9d0f1..cfe766445b 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -105,13 +105,20 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg database := controller.NewGroupDatabase(rdb, &config.LocalCacheConfig, groupDB, groupMemberDB, groupRequestDB, mgocli.GetTx(), grouphash.NewGroupHashFromGroupServer(&gs)) gs.db = database gs.user = userRpcClient - gs.notification = NewGroupNotificationSender(database, &msgRpcClient, &userRpcClient, config, func(ctx context.Context, userIDs []string) ([]notification.CommonUser, error) { - users, err := userRpcClient.GetUsersInfo(ctx, userIDs) - if err != nil { - return nil, err - } - return datautil.Slice(users, func(e *sdkws.UserInfo) notification.CommonUser { return e }), nil - }) + gs.notification = NewGroupNotificationSender( + database, + &msgRpcClient, + &userRpcClient, + &conversationRpcClient, + config, + func(ctx context.Context, userIDs []string) ([]notification.CommonUser, error) { + users, err := userRpcClient.GetUsersInfo(ctx, userIDs) + if err != nil { + return nil, err + } + return datautil.Slice(users, func(e *sdkws.UserInfo) notification.CommonUser { return e }), nil + }, + ) localcache.InitLocalCache(&config.LocalCacheConfig) gs.conversationRpcClient = conversationRpcClient gs.msgRpcClient = msgRpcClient @@ -453,7 +460,10 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite if err := s.conversationRpcClient.GroupChatFirstCreateConversation(ctx, req.GroupID, req.InvitedUserIDs); err != nil { return nil, err } - s.notification.MemberInvitedNotification(ctx, req.GroupID, req.Reason, req.InvitedUserIDs) + + if err = s.notification.MemberEnterNotification(ctx, req.GroupID, req.InvitedUserIDs...); err != nil { + return nil, err + } return &pbgroup.InviteUserToGroupResp{}, nil } @@ -829,7 +839,9 @@ func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbgroup if member == nil { log.ZDebug(ctx, "GroupApplicationResponse", "member is nil") } else { - s.notification.MemberEnterNotification(ctx, req.GroupID, req.FromUserID) + if err = s.notification.MemberEnterNotification(ctx, req.GroupID, req.FromUserID); err != nil { + return nil, err + } } case constant.GroupResponseRefuse: s.notification.GroupApplicationRejectedNotification(ctx, req) @@ -892,7 +904,9 @@ func (s *groupServer) JoinGroup(ctx context.Context, req *pbgroup.JoinGroupReq) if err := s.conversationRpcClient.GroupChatFirstCreateConversation(ctx, req.GroupID, []string{req.InviterUserID}); err != nil { return nil, err } - s.notification.MemberEnterNotification(ctx, req.GroupID, req.InviterUserID) + if err = s.notification.MemberEnterNotification(ctx, req.GroupID, req.InviterUserID); err != nil { + return nil, err + } s.webhookAfterJoinGroup(ctx, &s.config.WebhooksConfig.AfterJoinGroup, req) return &pbgroup.JoinGroupResp{}, nil diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index 9815167e9b..0be307d62f 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -21,6 +21,7 @@ import ( "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/open-im-server/v3/pkg/common/storage/versionctx" + "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/open-im-server/v3/pkg/rpcclient/notification" "github.com/openimsdk/open-im-server/v3/pkg/authverify" @@ -43,12 +44,21 @@ const ( adminReceiver ) -func NewGroupNotificationSender(db controller.GroupDatabase, msgRpcClient *rpcclient.MessageRpcClient, userRpcClient *rpcclient.UserRpcClient, config *Config, fn func(ctx context.Context, userIDs []string) ([]notification.CommonUser, error)) *GroupNotificationSender { +func NewGroupNotificationSender( + db controller.GroupDatabase, + msgRpcClient *rpcclient.MessageRpcClient, + userRpcClient *rpcclient.UserRpcClient, + conversationRpcClient *rpcclient.ConversationRpcClient, + config *Config, + fn func(ctx context.Context, userIDs []string) ([]notification.CommonUser, error), +) *GroupNotificationSender { return &GroupNotificationSender{ NotificationSender: rpcclient.NewNotificationSender(&config.NotificationConfig, rpcclient.WithRpcClient(msgRpcClient), rpcclient.WithUserRpcClient(userRpcClient)), getUsersInfo: fn, db: db, config: config, + + conversationRpcClient: conversationRpcClient, } } @@ -57,6 +67,8 @@ type GroupNotificationSender struct { getUsersInfo func(ctx context.Context, userIDs []string) ([]notification.CommonUser, error) db controller.GroupDatabase config *Config + + conversationRpcClient *rpcclient.ConversationRpcClient } func (g *GroupNotificationSender) PopulateGroupMember(ctx context.Context, members ...*model.GroupMember) error { @@ -494,50 +506,42 @@ func (g *GroupNotificationSender) MemberKickedNotification(ctx context.Context, g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.MemberKickedNotification, tips) } -func (g *GroupNotificationSender) MemberInvitedNotification(ctx context.Context, groupID, reason string, invitedUserIDList []string) { +func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, groupID string, entrantUserID ...string) error { var err error defer func() { if err != nil { log.ZError(ctx, stringutil.GetFuncName(1)+" failed", err) } }() - var group *sdkws.GroupInfo - group, err = g.getGroupInfo(ctx, groupID) - if err != nil { - return - } - var users []*sdkws.GroupMemberFullInfo - users, err = g.getGroupMembers(ctx, groupID, invitedUserIDList) - if err != nil { - return - } - tips := &sdkws.MemberInvitedTips{Group: group, InvitedUserList: users} - err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID) - g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) - g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberInvitedNotification, tips) -} - -func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, groupID string, entrantUserID string) { - var err error - defer func() { + if !g.config.RpcConfig.EnableHistoryForNewMembers { + conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID) + conversation, err := g.conversationRpcClient.GetConversationsByConversationID(ctx, []string{conversationID}) if err != nil { - log.ZError(ctx, stringutil.GetFuncName(1)+" failed", err) + return err } - }() + if len(conversation) == 0 { + return errs.New("group conversation not found").Wrap() + } + err = g.conversationRpcClient.SetConversationMaxSeq(ctx, entrantUserID, conversationID, conversation[0].MaxSeq) + if err != nil { + return err + } + } + var group *sdkws.GroupInfo group, err = g.getGroupInfo(ctx, groupID) if err != nil { - return + return err } - var user *sdkws.GroupMemberFullInfo - user, err = g.getGroupMember(ctx, groupID, entrantUserID) + users, err := g.getGroupMembers(ctx, groupID, entrantUserID) if err != nil { - return + return err } - tips := &sdkws.MemberEnterTips{Group: group, EntrantUser: user} + tips := &sdkws.MemberEnterTips{Group: group, EntrantUser: users} g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberEnterNotification, tips) + return nil } func (g *GroupNotificationSender) GroupDismissedNotification(ctx context.Context, tips *sdkws.GroupDismissedTips) { diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index c6c672eb8d..fc6d3001ff 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -258,7 +258,8 @@ type Group struct { ListenIP string `mapstructure:"listenIP"` Ports []int `mapstructure:"ports"` } `mapstructure:"rpc"` - Prometheus Prometheus `mapstructure:"prometheus"` + Prometheus Prometheus `mapstructure:"prometheus"` + EnableHistoryForNewMembers bool `mapstructure:"enableHistoryForNewMembers"` } type Msg struct { From 7e67d0923a563ee2a1aeb409df73929748ed2ac3 Mon Sep 17 00:00:00 2001 From: icey-yu <1186114839@qq.com> Date: Thu, 22 Aug 2024 18:00:57 +0800 Subject: [PATCH 2/4] feat: change name --- internal/rpc/group/notification.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index 0be307d62f..52f18879ae 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -538,7 +538,7 @@ func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, g if err != nil { return err } - tips := &sdkws.MemberEnterTips{Group: group, EntrantUser: users} + tips := &sdkws.MemberEnterTips{Group: group, EntrantUsers: users} g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberEnterNotification, tips) return nil From d5e7261f3ce41c15844a7ef31ed61c2e99484867 Mon Sep 17 00:00:00 2001 From: icey-yu <1186114839@qq.com> Date: Thu, 22 Aug 2024 18:04:00 +0800 Subject: [PATCH 3/4] refactor: move first create conversation --- internal/rpc/group/group.go | 9 --------- internal/rpc/group/notification.go | 4 ++++ 2 files changed, 4 insertions(+), 9 deletions(-) diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index cfe766445b..bb3bfcaeea 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -457,9 +457,6 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite if err := s.db.CreateGroup(ctx, nil, groupMembers); err != nil { return nil, err } - if err := s.conversationRpcClient.GroupChatFirstCreateConversation(ctx, req.GroupID, req.InvitedUserIDs); err != nil { - return nil, err - } if err = s.notification.MemberEnterNotification(ctx, req.GroupID, req.InvitedUserIDs...); err != nil { return nil, err @@ -832,9 +829,6 @@ func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbgroup } switch req.HandleResult { case constant.GroupResponseAgree: - if err := s.conversationRpcClient.GroupChatFirstCreateConversation(ctx, req.GroupID, []string{req.FromUserID}); err != nil { - return nil, err - } s.notification.GroupApplicationAcceptedNotification(ctx, req) if member == nil { log.ZDebug(ctx, "GroupApplicationResponse", "member is nil") @@ -901,9 +895,6 @@ func (s *groupServer) JoinGroup(ctx context.Context, req *pbgroup.JoinGroupReq) return nil, err } - if err := s.conversationRpcClient.GroupChatFirstCreateConversation(ctx, req.GroupID, []string{req.InviterUserID}); err != nil { - return nil, err - } if err = s.notification.MemberEnterNotification(ctx, req.GroupID, req.InviterUserID); err != nil { return nil, err } diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index 52f18879ae..bd0ebdb963 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -529,6 +529,10 @@ func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, g } } + if err := g.conversationRpcClient.GroupChatFirstCreateConversation(ctx, groupID, entrantUserID); err != nil { + return err + } + var group *sdkws.GroupInfo group, err = g.getGroupInfo(ctx, groupID) if err != nil { From fc469ad25782ea87b933a60c87791ce5225ac2b4 Mon Sep 17 00:00:00 2001 From: icey-yu <1186114839@qq.com> Date: Thu, 22 Aug 2024 18:37:28 +0800 Subject: [PATCH 4/4] fix: set min seq --- go.mod | 2 +- go.sum | 4 ++-- internal/rpc/conversation/conversaion.go | 8 ++++++++ internal/rpc/group/notification.go | 9 ++++----- pkg/rpcclient/conversation.go | 5 +++++ 5 files changed, 20 insertions(+), 8 deletions(-) diff --git a/go.mod b/go.mod index 6d501e0525..1d7c648f7e 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/gorilla/websocket v1.5.1 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/mitchellh/mapstructure v1.5.0 - github.com/openimsdk/protocol v0.0.72-alpha.1 + github.com/openimsdk/protocol v0.0.72-alpha.2 github.com/openimsdk/tools v0.0.49-alpha.55 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 diff --git a/go.sum b/go.sum index 1c011e42e5..5e0874723d 100644 --- a/go.sum +++ b/go.sum @@ -319,8 +319,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y= github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM= github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y= github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= -github.com/openimsdk/protocol v0.0.72-alpha.1 h1:qaNV6+mBme2NC9lXWAScgKj5gjAcihKixKzxTndDzvw= -github.com/openimsdk/protocol v0.0.72-alpha.1/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= +github.com/openimsdk/protocol v0.0.72-alpha.2 h1:H5IcoAR4jTzJ7uvmFmc6/6rPOKEGRLWnu+qre8B5hk0= +github.com/openimsdk/protocol v0.0.72-alpha.2/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= github.com/openimsdk/tools v0.0.49-alpha.55 h1:KPgC53oqiwZYssLKljhtXbWXifMlTj2SSQEusj4Uf4k= github.com/openimsdk/tools v0.0.49-alpha.55/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4= github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4= diff --git a/internal/rpc/conversation/conversaion.go b/internal/rpc/conversation/conversaion.go index 2fce2698ab..8ee156829a 100644 --- a/internal/rpc/conversation/conversaion.go +++ b/internal/rpc/conversation/conversaion.go @@ -402,6 +402,14 @@ func (c *conversationServer) SetConversationMaxSeq(ctx context.Context, req *pbc return &pbconversation.SetConversationMaxSeqResp{}, nil } +func (c *conversationServer) SetConversationMinSeq(ctx context.Context, req *pbconversation.SetConversationMinSeqReq) (*pbconversation.SetConversationMinSeqResp, error) { + if err := c.conversationDatabase.UpdateUsersConversationField(ctx, req.OwnerUserID, req.ConversationID, + map[string]any{"min_seq": req.MinSeq}); err != nil { + return nil, err + } + return &pbconversation.SetConversationMinSeqResp{}, nil +} + func (c *conversationServer) GetConversationIDs(ctx context.Context, req *pbconversation.GetConversationIDsReq) (*pbconversation.GetConversationIDsResp, error) { conversationIDs, err := c.conversationDatabase.GetConversationIDs(ctx, req.UserID) if err != nil { diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index bd0ebdb963..72ca9b8130 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -59,6 +59,7 @@ func NewGroupNotificationSender( config: config, conversationRpcClient: conversationRpcClient, + msgRpcClient: msgRpcClient, } } @@ -69,6 +70,7 @@ type GroupNotificationSender struct { config *Config conversationRpcClient *rpcclient.ConversationRpcClient + msgRpcClient *rpcclient.MessageRpcClient } func (g *GroupNotificationSender) PopulateGroupMember(ctx context.Context, members ...*model.GroupMember) error { @@ -516,14 +518,11 @@ func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, g if !g.config.RpcConfig.EnableHistoryForNewMembers { conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID) - conversation, err := g.conversationRpcClient.GetConversationsByConversationID(ctx, []string{conversationID}) + maxSeq, err := g.msgRpcClient.GetConversationMaxSeq(ctx, conversationID) if err != nil { return err } - if len(conversation) == 0 { - return errs.New("group conversation not found").Wrap() - } - err = g.conversationRpcClient.SetConversationMaxSeq(ctx, entrantUserID, conversationID, conversation[0].MaxSeq) + err = g.conversationRpcClient.SetConversationMinSeq(ctx, entrantUserID, conversationID, maxSeq) if err != nil { return err } diff --git a/pkg/rpcclient/conversation.go b/pkg/rpcclient/conversation.go index 8f95f86a6c..ccca856194 100644 --- a/pkg/rpcclient/conversation.go +++ b/pkg/rpcclient/conversation.go @@ -77,6 +77,11 @@ func (c *ConversationRpcClient) SetConversationMaxSeq(ctx context.Context, owner return err } +func (c *ConversationRpcClient) SetConversationMinSeq(ctx context.Context, ownerUserIDs []string, conversationID string, minSeq int64) error { + _, err := c.Client.SetConversationMinSeq(ctx, &pbconversation.SetConversationMinSeqReq{OwnerUserID: ownerUserIDs, ConversationID: conversationID, MinSeq: minSeq}) + return err +} + func (c *ConversationRpcClient) SetConversations(ctx context.Context, userIDs []string, conversation *pbconversation.ConversationReq) error { _, err := c.Client.SetConversations(ctx, &pbconversation.SetConversationsReq{UserIDs: userIDs, Conversation: conversation}) return err