diff --git a/config/openim-push.yml b/config/openim-push.yml index a1abfcf88..9384008a0 100644 --- a/config/openim-push.yml +++ b/config/openim-push.yml @@ -23,7 +23,9 @@ geTui: channelID: '' channelName: '' fcm: - serviceAccount: "x.json" + # Prioritize using file paths. If the file path is empty, use URL + filePath: "" # File path is concatenated with the parameters passed in through - c(`mage` default pass in `config/`) and filePath. + authURL: "" # Must start with https or http. jpns: appKey: '' masterSecret: '' diff --git a/config/redis.yml b/config/redis.yml index 404d18953..87abed0e1 100644 --- a/config/redis.yml +++ b/config/redis.yml @@ -1,4 +1,4 @@ -address: [ 172.16.8.48:16379 ] +address: [ localhost:16379 ] username: '' password: openIM123 clusterMode: false diff --git a/go.mod b/go.mod index 4f4cd4b68..d00f0b739 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect github.com/mitchellh/mapstructure v1.5.0 - github.com/openimsdk/protocol v0.0.69-alpha.9 + github.com/openimsdk/protocol v0.0.69-alpha.15 github.com/openimsdk/tools v0.0.49-alpha.25 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 @@ -177,4 +177,6 @@ require ( gopkg.in/ini.v1 v1.67.0 // indirect ) -//replace github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol +//replace ( +// github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol +//) diff --git a/go.sum b/go.sum index 5a72e28f2..4c05500bd 100644 --- a/go.sum +++ b/go.sum @@ -270,8 +270,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.13 h1:xLDe/moqgWpRoptHzI4packAWzs4C16b+sVY+txNJp0= github.com/openimsdk/gomake v0.0.13/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= -github.com/openimsdk/protocol v0.0.69-alpha.9 h1:Mh1upsxwhWs1y65cfIN2XuRsiKEk4MfpYfA9XrvtX24= -github.com/openimsdk/protocol v0.0.69-alpha.9/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= +github.com/openimsdk/protocol v0.0.69-alpha.15 h1:iiVROC2nXTc9qPOSfE7NZckQAVCZiDxBm9yY36ULdyw= +github.com/openimsdk/protocol v0.0.69-alpha.15/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= github.com/openimsdk/tools v0.0.49-alpha.25 h1:OpRPwDZ2xWX7Zj5kyfZhryu/NfZTrsRVr2GFwu1HQHI= github.com/openimsdk/tools v0.0.49-alpha.25/go.mod h1:rwsFI1G/nBHNfiNapbven41akRDPBbH4df0Cgy6xueU= github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4= diff --git a/internal/api/friend.go b/internal/api/friend.go index 6912fdbbb..9266fe75c 100644 --- a/internal/api/friend.go +++ b/internal/api/friend.go @@ -102,3 +102,7 @@ func (o *FriendApi) GetIncrementalFriends(c *gin.Context) { func (o *FriendApi) GetIncrementalBlacks(c *gin.Context) { a2r.Call(relation.FriendClient.GetIncrementalBlacks, o.Client, c) } + +func (o *FriendApi) GetFullFriendUserIDs(c *gin.Context) { + a2r.Call(relation.FriendClient.GetFullFriendUserIDs, o.Client, c) +} diff --git a/internal/api/group.go b/internal/api/group.go index 0bf61c787..91992004c 100644 --- a/internal/api/group.go +++ b/internal/api/group.go @@ -187,10 +187,10 @@ func (o *GroupApi) GetIncrementalGroupMemberBatch(c *gin.Context) { apiresp.GinSuccess(c, resp) } -func (o *GroupApi) GetIncrementalGroupMemberUserIDs(c *gin.Context) { - a2r.Call(group.GroupClient.GetIncrementalGroupMemberUserIDs, o.Client, c) +func (o *GroupApi) GetFullGroupMemberUserIDs(c *gin.Context) { + a2r.Call(group.GroupClient.GetFullGroupMemberUserIDs, o.Client, c) } -func (o *GroupApi) GetIncrementalJoinGroupIDs(c *gin.Context) { - a2r.Call(group.GroupClient.GetIncrementalJoinGroupIDs, o.Client, c) +func (o *GroupApi) GetFullJoinGroupIDs(c *gin.Context) { + a2r.Call(group.GroupClient.GetFullJoinGroupIDs, o.Client, c) } diff --git a/internal/api/router.go b/internal/api/router.go index 69c026ffd..0f46f26ba 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -89,6 +89,7 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En friendRouterGroup.POST("/get_specified_friends_info", f.GetSpecifiedFriendsInfo) friendRouterGroup.POST("/update_friends", f.UpdateFriends) friendRouterGroup.POST("/get_incremental_friends", f.GetIncrementalFriends) + friendRouterGroup.POST("/get_full_friend_user_ids", f.GetFullFriendUserIDs) } g := NewGroupApi(*groupRpc) groupRouterGroup := r.Group("/group") @@ -120,8 +121,8 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En groupRouterGroup.POST("/get_incremental_join_group", g.GetIncrementalJoinGroup) groupRouterGroup.POST("/get_incremental_group_member", g.GetIncrementalGroupMember) groupRouterGroup.POST("/get_incremental_group_member_batch", g.GetIncrementalGroupMemberBatch) - groupRouterGroup.POST("/get_incremental_group_member_user_ids", g.GetIncrementalGroupMemberUserIDs) - groupRouterGroup.POST("/get_incremental_join_group_ids", g.GetIncrementalJoinGroupIDs) + groupRouterGroup.POST("/get_full_group_member_user_ids", g.GetFullGroupMemberUserIDs) + groupRouterGroup.POST("/get_full_join_group_ids", g.GetFullJoinGroupIDs) } // certificate authRouterGroup := r.Group("/auth") diff --git a/internal/push/offlinepush/fcm/push.go b/internal/push/offlinepush/fcm/push.go index ec973008e..f015ca4e5 100644 --- a/internal/push/offlinepush/fcm/push.go +++ b/internal/push/offlinepush/fcm/push.go @@ -16,8 +16,11 @@ package fcm import ( "context" + "fmt" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" + "github.com/openimsdk/tools/utils/httputil" "path/filepath" + "strings" firebase "firebase.google.com/go" "firebase.google.com/go/messaging" @@ -40,13 +43,25 @@ type Fcm struct { // NewClient initializes a new FCM client using the Firebase Admin SDK. // It requires the FCM service account credentials file located within the project's configuration directory. -func NewClient(pushConf *config.Push, cache cache.ThirdCache) (*Fcm, error) { - projectRoot, err := config.GetProjectRoot() - if err != nil { - return nil, err +func NewClient(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (*Fcm, error) { + var opt option.ClientOption + switch { + case len(pushConf.FCM.FilePath) != 0: + // with file path + credentialsFilePath := filepath.Join(fcmConfigPath, pushConf.FCM.FilePath) + opt = option.WithCredentialsFile(credentialsFilePath) + case len(pushConf.FCM.AuthURL) != 0: + // with authentication URL + client := httputil.NewHTTPClient(httputil.NewClientConfig()) + resp, err := client.Get(pushConf.FCM.AuthURL) + if err != nil { + return nil, err + } + opt = option.WithCredentialsJSON(resp) + default: + return nil, errs.New("no FCM config").Wrap() } - credentialsFilePath := filepath.Join(projectRoot, "config", pushConf.FCM.ServiceAccount) - opt := option.WithCredentialsFile(credentialsFilePath) + fcmApp, err := firebase.NewApp(context.Background(), nil, opt) if err != nil { return nil, errs.Wrap(err) @@ -56,7 +71,6 @@ func NewClient(pushConf *config.Push, cache cache.ThirdCache) (*Fcm, error) { if err != nil { return nil, errs.Wrap(err) } - return &Fcm{fcmMsgCli: fcmMsgClient, cache: cache}, nil } @@ -79,6 +93,8 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, notification.Body = content notification.Title = title var messages []*messaging.Message + var sendErrBuilder strings.Builder + var msgErrBuilder strings.Builder for userID, personTokens := range allTokens { apns := &messaging.APNSConfig{Payload: &messaging.APNSPayload{Aps: &messaging.Aps{Sound: opts.IOSPushSound}}} messageCount := len(messages) @@ -86,9 +102,21 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, response, err := f.fcmMsgCli.SendAll(ctx, messages) if err != nil { Fail = Fail + messageCount + // Record push error + sendErrBuilder.WriteString(err.Error()) + sendErrBuilder.WriteByte('.') } else { Success = Success + response.SuccessCount Fail = Fail + response.FailureCount + if response.FailureCount != 0 { + // Record message error + for i := range response.Responses { + if !response.Responses[i].Success { + msgErrBuilder.WriteString(response.Responses[i].Error.Error()) + msgErrBuilder.WriteByte('.') + } + } + } } messages = messages[0:0] } @@ -134,5 +162,9 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, Fail = Fail + response.FailureCount } } + if Fail != 0 { + return errs.New(fmt.Sprintf("%d message send failed;send err:%s;message err:%s", + Fail, sendErrBuilder.String(), msgErrBuilder.String())).Wrap() + } return nil } diff --git a/internal/push/offlinepush/offlinepusher.go b/internal/push/offlinepush/offlinepusher.go index 8dc8a0bc6..9aa6625de 100644 --- a/internal/push/offlinepush/offlinepusher.go +++ b/internal/push/offlinepush/offlinepusher.go @@ -36,13 +36,13 @@ type OfflinePusher interface { Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error } -func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache) (OfflinePusher, error) { +func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (OfflinePusher, error) { var offlinePusher OfflinePusher switch pushConf.Enable { case geTUI: offlinePusher = getui.NewClient(pushConf, cache) case firebase: - return fcm.NewClient(pushConf, cache) + return fcm.NewClient(pushConf, cache, fcmConfigPath) case jPush: offlinePusher = jpush.NewClient(pushConf) default: diff --git a/internal/push/push.go b/internal/push/push.go index c7e245dfe..1a04bbea2 100644 --- a/internal/push/push.go +++ b/internal/push/push.go @@ -29,6 +29,7 @@ type Config struct { WebhooksConfig config.Webhooks LocalCacheConfig config.LocalCache Discovery config.Discovery + FcmConfigPath string } func (p pushServer) PushMsg(ctx context.Context, req *pbpush.PushMsgReq) (*pbpush.PushMsgResp, error) { @@ -50,7 +51,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg return err } cacheModel := redis.NewThirdCache(rdb) - offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel) + offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel, config.FcmConfigPath) if err != nil { return err } diff --git a/internal/rpc/auth/auth.go b/internal/rpc/auth/auth.go index 6270b39b3..320fb1d52 100644 --- a/internal/rpc/auth/auth.go +++ b/internal/rpc/auth/auth.go @@ -16,6 +16,7 @@ package auth import ( "context" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" redis2 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" "github.com/openimsdk/tools/db/redisutil" @@ -32,7 +33,6 @@ import ( "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/tokenverify" "google.golang.org/grpc" ) @@ -153,21 +153,19 @@ func (s *authServer) ForceLogout(ctx context.Context, req *pbauth.ForceLogoutReq if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil { return nil, err } - if err := s.forceKickOff(ctx, req.UserID, req.PlatformID, mcontext.GetOperationID(ctx)); err != nil { + if err := s.forceKickOff(ctx, req.UserID, req.PlatformID); err != nil { return nil, err } return &pbauth.ForceLogoutResp{}, nil } -func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID int32, operationID string) error { +func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID int32) error { conns, err := s.RegisterCenter.GetConns(ctx, s.config.Share.RpcRegisterName.MessageGateway) if err != nil { return err } for _, v := range conns { log.ZDebug(ctx, "forceKickOff", "conn", v.Target()) - } - for _, v := range conns { client := msggateway.NewMsgGatewayClient(v) kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID} _, err := client.KickUserOffline(ctx, kickReq) @@ -175,8 +173,24 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID log.ZError(ctx, "forceKickOff", err, "kickReq", kickReq) } } + + m, err := s.authDatabase.GetTokensWithoutError(ctx, userID, int(platformID)) + if err != nil && err != redis.Nil { + return err + } + for k := range m { + m[k] = constant.KickedToken + log.ZDebug(ctx, "set token map is ", "token map", m, "userID", + userID, "token", k) + + err = s.authDatabase.SetTokenMapByUidPid(ctx, userID, int(platformID), m) + if err != nil { + return err + } + } return nil } + func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) { m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID)) if err != nil && err != redis.Nil { diff --git a/internal/rpc/friend/friend.go b/internal/rpc/friend/friend.go index 12107125c..622e19f42 100644 --- a/internal/rpc/friend/friend.go +++ b/internal/rpc/friend/friend.go @@ -41,7 +41,7 @@ import ( ) type friendServer struct { - friendDatabase controller.FriendDatabase + db controller.FriendDatabase blackDatabase controller.BlackDatabase userRpcClient *rpcclient.UserRpcClient notificationSender *FriendNotificationSender @@ -102,7 +102,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg // Register Friend server with refactored MongoDB and Redis integrations relation.RegisterFriendServer(server, &friendServer{ - friendDatabase: controller.NewFriendDatabase( + db: controller.NewFriendDatabase( friendMongoDB, friendRequestMongoDB, redis.NewFriendCacheRedis(rdb, &config.LocalCacheConfig, friendMongoDB, redis.GetRocksCacheOptions()), @@ -139,14 +139,14 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *relation.Apply return nil, err } - in1, in2, err := s.friendDatabase.CheckIn(ctx, req.FromUserID, req.ToUserID) + in1, in2, err := s.db.CheckIn(ctx, req.FromUserID, req.ToUserID) if err != nil { return nil, err } if in1 && in2 { return nil, servererrs.ErrRelationshipAlready.WrapMsg("already friends has f") } - if err = s.friendDatabase.AddFriendRequest(ctx, req.FromUserID, req.ToUserID, req.ReqMsg, req.Ex); err != nil { + if err = s.db.AddFriendRequest(ctx, req.FromUserID, req.ToUserID, req.ReqMsg, req.Ex); err != nil { return nil, err } s.notificationSender.FriendApplicationAddNotification(ctx, req) @@ -173,7 +173,7 @@ func (s *friendServer) ImportFriends(ctx context.Context, req *relation.ImportFr return nil, err } - if err := s.friendDatabase.BecomeFriends(ctx, req.OwnerUserID, req.FriendUserIDs, constant.BecomeFriendByImport); err != nil { + if err := s.db.BecomeFriends(ctx, req.OwnerUserID, req.FriendUserIDs, constant.BecomeFriendByImport); err != nil { return nil, err } for _, userID := range req.FriendUserIDs { @@ -205,7 +205,7 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *relation.Res if err := s.webhookBeforeAddFriendAgree(ctx, &s.config.WebhooksConfig.BeforeAddFriendAgree, req); err != nil && err != servererrs.ErrCallbackContinue { return nil, err } - err := s.friendDatabase.AgreeFriendRequest(ctx, &friendRequest) + err := s.db.AgreeFriendRequest(ctx, &friendRequest) if err != nil { return nil, err } @@ -213,7 +213,7 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *relation.Res return resp, nil } if req.HandleResult == constant.FriendResponseRefuse { - err := s.friendDatabase.RefuseFriendRequest(ctx, &friendRequest) + err := s.db.RefuseFriendRequest(ctx, &friendRequest) if err != nil { return nil, err } @@ -229,11 +229,11 @@ func (s *friendServer) DeleteFriend(ctx context.Context, req *relation.DeleteFri if err := s.userRpcClient.Access(ctx, req.OwnerUserID); err != nil { return nil, err } - _, err = s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, []string{req.FriendUserID}) + _, err = s.db.FindFriendsWithError(ctx, req.OwnerUserID, []string{req.FriendUserID}) if err != nil { return nil, err } - if err := s.friendDatabase.Delete(ctx, req.OwnerUserID, []string{req.FriendUserID}); err != nil { + if err := s.db.Delete(ctx, req.OwnerUserID, []string{req.FriendUserID}); err != nil { return nil, err } s.notificationSender.FriendDeletedNotification(ctx, req) @@ -250,11 +250,11 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *relation.SetFri if err := s.userRpcClient.Access(ctx, req.OwnerUserID); err != nil { return nil, err } - _, err = s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, []string{req.FriendUserID}) + _, err = s.db.FindFriendsWithError(ctx, req.OwnerUserID, []string{req.FriendUserID}) if err != nil { return nil, err } - if err := s.friendDatabase.UpdateRemark(ctx, req.OwnerUserID, req.FriendUserID, req.Remark); err != nil { + if err := s.db.UpdateRemark(ctx, req.OwnerUserID, req.FriendUserID, req.Remark); err != nil { return nil, err } s.webhookAfterSetFriendRemark(ctx, &s.config.WebhooksConfig.AfterSetFriendRemark, req) @@ -281,7 +281,7 @@ func (s *friendServer) getFriend(ctx context.Context, ownerUserID string, friend if len(friendUserIDs) == 0 { return nil, nil } - friends, err := s.friendDatabase.FindFriendsWithError(ctx, ownerUserID, friendUserIDs) + friends, err := s.db.FindFriendsWithError(ctx, ownerUserID, friendUserIDs) if err != nil { return nil, err } @@ -292,7 +292,7 @@ func (s *friendServer) getFriend(ctx context.Context, ownerUserID string, friend func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context, req *relation.GetDesignatedFriendsApplyReq, ) (resp *relation.GetDesignatedFriendsApplyResp, err error) { - friendRequests, err := s.friendDatabase.FindBothFriendRequests(ctx, req.FromUserID, req.ToUserID) + friendRequests, err := s.db.FindBothFriendRequests(ctx, req.FromUserID, req.ToUserID) if err != nil { return nil, err } @@ -309,7 +309,7 @@ func (s *friendServer) GetPaginationFriendsApplyTo(ctx context.Context, req *rel if err := s.userRpcClient.Access(ctx, req.UserID); err != nil { return nil, err } - total, friendRequests, err := s.friendDatabase.PageFriendRequestToMe(ctx, req.UserID, req.Pagination) + total, friendRequests, err := s.db.PageFriendRequestToMe(ctx, req.UserID, req.Pagination) if err != nil { return nil, err } @@ -327,7 +327,7 @@ func (s *friendServer) GetPaginationFriendsApplyFrom(ctx context.Context, req *r if err := s.userRpcClient.Access(ctx, req.UserID); err != nil { return nil, err } - total, friendRequests, err := s.friendDatabase.PageFriendRequestFromMe(ctx, req.UserID, req.Pagination) + total, friendRequests, err := s.db.PageFriendRequestFromMe(ctx, req.UserID, req.Pagination) if err != nil { return nil, err } @@ -342,7 +342,7 @@ func (s *friendServer) GetPaginationFriendsApplyFrom(ctx context.Context, req *r // ok. func (s *friendServer) IsFriend(ctx context.Context, req *relation.IsFriendReq) (resp *relation.IsFriendResp, err error) { resp = &relation.IsFriendResp{} - resp.InUser1Friends, resp.InUser2Friends, err = s.friendDatabase.CheckIn(ctx, req.UserID1, req.UserID2) + resp.InUser1Friends, resp.InUser2Friends, err = s.db.CheckIn(ctx, req.UserID1, req.UserID2) if err != nil { return nil, err } @@ -353,7 +353,7 @@ func (s *friendServer) GetPaginationFriends(ctx context.Context, req *relation.G if err := s.userRpcClient.Access(ctx, req.UserID); err != nil { return nil, err } - total, friends, err := s.friendDatabase.PageOwnerFriends(ctx, req.UserID, req.Pagination) + total, friends, err := s.db.PageOwnerFriends(ctx, req.UserID, req.Pagination) if err != nil { return nil, err } @@ -371,7 +371,7 @@ func (s *friendServer) GetFriendIDs(ctx context.Context, req *relation.GetFriend return nil, err } resp = &relation.GetFriendIDsResp{} - resp.FriendIDs, err = s.friendDatabase.FindFriendUserIDs(ctx, req.UserID) + resp.FriendIDs, err = s.db.FindFriendUserIDs(ctx, req.UserID) if err != nil { return nil, err } @@ -389,7 +389,7 @@ func (s *friendServer) GetSpecifiedFriendsInfo(ctx context.Context, req *relatio if err != nil { return nil, err } - friends, err := s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, req.UserIDList) + friends, err := s.db.FindFriendsWithError(ctx, req.OwnerUserID, req.UserIDList) if err != nil { return nil, err } @@ -453,7 +453,7 @@ func (s *friendServer) UpdateFriends( return nil, errs.ErrArgs.WrapMsg("friendIDList repeated") } - _, err := s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs) + _, err := s.db.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs) if err != nil { return nil, err } @@ -469,7 +469,7 @@ func (s *friendServer) UpdateFriends( if req.Ex != nil { val["ex"] = req.Ex.Value } - if err = s.friendDatabase.UpdateFriends(ctx, req.OwnerUserID, req.FriendUserIDs, val); err != nil { + if err = s.db.UpdateFriends(ctx, req.OwnerUserID, req.FriendUserIDs, val); err != nil { return nil, err } diff --git a/internal/rpc/friend/sync.go b/internal/rpc/friend/sync.go index cddd23f89..faaa987f2 100644 --- a/internal/rpc/friend/sync.go +++ b/internal/rpc/friend/sync.go @@ -2,6 +2,7 @@ package friend import ( "context" + "github.com/openimsdk/open-im-server/v3/pkg/util/hashutil" "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" @@ -10,6 +11,27 @@ import ( "github.com/openimsdk/protocol/relation" ) +func (s *friendServer) GetFullFriendUserIDs(ctx context.Context, req *relation.GetFullFriendUserIDsReq) (*relation.GetFullFriendUserIDsResp, error) { + vl, err := s.db.FindMaxFriendVersionCache(ctx, req.UserID) + if err != nil { + return nil, err + } + userIDs, err := s.db.FindFriendUserIDs(ctx, req.UserID) + if err != nil { + return nil, err + } + idHash := hashutil.IdHash(userIDs) + if req.IdHash == idHash { + userIDs = nil + } + return &relation.GetFullFriendUserIDsResp{ + Version: idHash, + VersionID: vl.ID.Hex(), + Equal: req.IdHash == idHash, + UserIDs: userIDs, + }, nil +} + func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.GetIncrementalFriendsReq) (*relation.GetIncrementalFriendsResp, error) { if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil { return nil, err @@ -19,8 +41,8 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation. VersionKey: req.UserID, VersionID: req.VersionID, VersionNumber: req.Version, - Version: s.friendDatabase.FindFriendIncrVersion, - CacheMaxVersion: s.friendDatabase.FindMaxFriendVersionCache, + Version: s.db.FindFriendIncrVersion, + CacheMaxVersion: s.db.FindMaxFriendVersionCache, Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) { return s.getFriend(ctx, req.UserID, ids) }, diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index 13e593c71..5d385d8c1 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -289,10 +289,12 @@ func (g *GroupNotificationSender) fillOpUser(ctx context.Context, opUser **sdkws return nil } -func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint64, collName string, id string) { - for _, coll := range versionctx.GetVersionLog(ctx).Get() { +func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string) { + versions := versionctx.GetVersionLog(ctx).Get() + for _, coll := range versions { if coll.Name == collName && coll.Doc.DID == id { *version = uint64(coll.Doc.Version) + *versionID = coll.Doc.ID.Hex() return } } @@ -308,7 +310,7 @@ func (g *GroupNotificationSender) GroupCreatedNotification(ctx context.Context, if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupCreatedNotification, tips) } @@ -322,6 +324,7 @@ func (g *GroupNotificationSender) GroupInfoSetNotification(ctx context.Context, if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetNotification, tips, rpcclient.WithRpcGetUserName()) } @@ -335,6 +338,7 @@ func (g *GroupNotificationSender) GroupInfoSetNameNotification(ctx context.Conte if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetNameNotification, tips) } @@ -348,6 +352,7 @@ func (g *GroupNotificationSender) GroupInfoSetAnnouncementNotification(ctx conte if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetAnnouncementNotification, tips, rpcclient.WithRpcGetUserName()) } @@ -392,7 +397,7 @@ func (g *GroupNotificationSender) MemberQuitNotification(ctx context.Context, me return } tips := &sdkws.MemberQuitTips{Group: group, QuitUser: member} - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, member.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, member.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), member.GroupID, constant.MemberQuitNotification, tips) } @@ -480,7 +485,7 @@ func (g *GroupNotificationSender) GroupOwnerTransferredNotification(ctx context. if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, req.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, req.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupOwnerTransferredNotification, tips) } @@ -494,7 +499,7 @@ func (g *GroupNotificationSender) MemberKickedNotification(ctx context.Context, if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.MemberKickedNotification, tips) } @@ -518,7 +523,7 @@ func (g *GroupNotificationSender) MemberInvitedNotification(ctx context.Context, } tips := &sdkws.MemberInvitedTips{Group: group, InvitedUserList: users} err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID) - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, 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) } @@ -540,7 +545,7 @@ func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, g return } tips := &sdkws.MemberEnterTips{Group: group, EntrantUser: user} - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberEnterNotification, tips) } @@ -581,7 +586,7 @@ func (g *GroupNotificationSender) GroupMemberMutedNotification(ctx context.Conte if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberMutedNotification, tips) } @@ -606,7 +611,7 @@ func (g *GroupNotificationSender) GroupMemberCancelMutedNotification(ctx context if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberCancelMutedNotification, tips) } @@ -634,6 +639,7 @@ func (g *GroupNotificationSender) GroupMutedNotification(ctx context.Context, gr if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, groupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMutedNotification, tips) } @@ -661,6 +667,7 @@ func (g *GroupNotificationSender) GroupCancelMutedNotification(ctx context.Conte if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, groupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupCancelMutedNotification, tips) } @@ -685,7 +692,7 @@ func (g *GroupNotificationSender) GroupMemberInfoSetNotification(ctx context.Con if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberInfoSetNotification, tips) } @@ -709,7 +716,7 @@ func (g *GroupNotificationSender) GroupMemberSetToAdminNotification(ctx context. if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToAdminNotification, tips) } @@ -734,6 +741,6 @@ func (g *GroupNotificationSender) GroupMemberSetToOrdinaryUserNotification(ctx c if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { return } - g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID) + g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToOrdinaryUserNotification, tips) } diff --git a/internal/rpc/group/sync.go b/internal/rpc/group/sync.go index 4bd34eb4d..3793d0c7e 100644 --- a/internal/rpc/group/sync.go +++ b/internal/rpc/group/sync.go @@ -2,26 +2,18 @@ package group import ( "context" - "crypto/md5" - "encoding/binary" - "encoding/json" "github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" "github.com/openimsdk/open-im-server/v3/pkg/authverify" + "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" + "github.com/openimsdk/open-im-server/v3/pkg/util/hashutil" + "github.com/openimsdk/protocol/constant" pbgroup "github.com/openimsdk/protocol/group" "github.com/openimsdk/protocol/sdkws" + "slices" ) -func (s *groupServer) idHash(ids []string) uint64 { - if len(ids) == 0 { - return 0 - } - data, _ := json.Marshal(ids) - sum := md5.Sum(data) - return binary.BigEndian.Uint64(sum[:]) -} - -func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetIncrementalGroupMemberUserIDsReq) (*pbgroup.GetIncrementalGroupMemberUserIDsResp, error) { +func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) { vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID) if err != nil { return nil, err @@ -30,11 +22,11 @@ func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req if err != nil { return nil, err } - idHash := s.idHash(userIDs) + idHash := hashutil.IdHash(userIDs) if req.IdHash == idHash { userIDs = nil } - return &pbgroup.GetIncrementalGroupMemberUserIDsResp{ + return &pbgroup.GetFullGroupMemberUserIDsResp{ Version: idHash, VersionID: vl.ID.Hex(), Equal: req.IdHash == idHash, @@ -42,7 +34,7 @@ func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req }, nil } -func (s *groupServer) GetIncrementalJoinGroupIDs(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupIDsReq) (*pbgroup.GetIncrementalJoinGroupIDsResp, error) { +func (s *groupServer) GetFullJoinGroupIDs(ctx context.Context, req *pbgroup.GetFullJoinGroupIDsReq) (*pbgroup.GetFullJoinGroupIDsResp, error) { vl, err := s.db.FindMaxJoinGroupVersionCache(ctx, req.UserID) if err != nil { return nil, err @@ -51,11 +43,11 @@ func (s *groupServer) GetIncrementalJoinGroupIDs(ctx context.Context, req *pbgro if err != nil { return nil, err } - idHash := s.idHash(groupIDs) + idHash := hashutil.IdHash(groupIDs) if req.IdHash == idHash { groupIDs = nil } - return &pbgroup.GetIncrementalJoinGroupIDsResp{ + return &pbgroup.GetFullJoinGroupIDsResp{ Version: idHash, VersionID: vl.ID.Hex(), Equal: req.IdHash == idHash, @@ -64,12 +56,34 @@ func (s *groupServer) GetIncrementalJoinGroupIDs(ctx context.Context, req *pbgro } func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgroup.GetIncrementalGroupMemberReq) (*pbgroup.GetIncrementalGroupMemberResp, error) { + group, err := s.db.TakeGroup(ctx, req.GroupID) + if err != nil { + return nil, err + } + if group.Status == constant.GroupStatusDismissed { + return nil, servererrs.ErrDismissedAlready.Wrap() + } + var hasGroupUpdate bool opt := incrversion.Option[*sdkws.GroupMemberFullInfo, pbgroup.GetIncrementalGroupMemberResp]{ - Ctx: ctx, - VersionKey: req.GroupID, - VersionID: req.VersionID, - VersionNumber: req.Version, - Version: s.db.FindMemberIncrVersion, + Ctx: ctx, + VersionKey: req.GroupID, + VersionID: req.VersionID, + VersionNumber: req.Version, + Version: func(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) { + vl, err := s.db.FindMemberIncrVersion(ctx, groupID, version, limit) + if err != nil { + return nil, err + } + vl.Logs = slices.DeleteFunc(vl.Logs, func(elem model.VersionLogElem) bool { + if elem.EID == "" { + vl.LogLen-- + hasGroupUpdate = true + return true + } + return false + }) + return vl, nil + }, CacheMaxVersion: s.db.FindMaxGroupMemberVersionCache, Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) { return s.getGroupMembersInfo(ctx, req.GroupID, ids) @@ -86,7 +100,22 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou } }, } - return opt.Build() + resp, err := opt.Build() + if err != nil { + return nil, err + } + if resp.Full || hasGroupUpdate { + count, err := s.db.FindGroupMemberNum(ctx, group.GroupID) + if err != nil { + return nil, err + } + owner, err := s.db.TakeGroupOwner(ctx, group.GroupID) + if err != nil { + return nil, err + } + resp.Group = s.groupDB2PB(group, owner.UserID, count) + } + return resp, nil } func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupReq) (*pbgroup.GetIncrementalJoinGroupResp, error) { diff --git a/pkg/common/cmd/push.go b/pkg/common/cmd/push.go index 3e7c4c249..6e6014021 100644 --- a/pkg/common/cmd/push.go +++ b/pkg/common/cmd/push.go @@ -47,6 +47,7 @@ func NewPushRpcCmd() *PushRpcCmd { ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap)) ret.ctx = context.WithValue(context.Background(), "version", config.Version) ret.Command.RunE = func(cmd *cobra.Command, args []string) error { + ret.pushConfig.FcmConfigPath = ret.ConfigPath() return ret.runE() } return ret diff --git a/pkg/common/cmd/root.go b/pkg/common/cmd/root.go index 900281367..08bb6d064 100644 --- a/pkg/common/cmd/root.go +++ b/pkg/common/cmd/root.go @@ -31,6 +31,11 @@ type RootCmd struct { prometheusPort int log config.Log index int + configPath string +} + +func (r *RootCmd) ConfigPath() string { + return r.configPath } func (r *RootCmd) Index() int { @@ -153,6 +158,7 @@ func (r *RootCmd) getFlag(cmd *cobra.Command) (string, int, error) { if err != nil { return "", 0, errs.Wrap(err) } + r.configPath = configDirectory index, err := cmd.Flags().GetInt(FlagTransferIndex) if err != nil { return "", 0, errs.Wrap(err) diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index 5313c196a..6260dc00f 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -202,7 +202,8 @@ type Push struct { ChannelName string `mapstructure:"channelName"` } `mapstructure:"geTui"` FCM struct { - ServiceAccount string `mapstructure:"serviceAccount"` + FilePath string `mapstructure:"filePath"` + AuthURL string `mapstructure:"authURL"` } `mapstructure:"fcm"` JPNS struct { AppKey string `mapstructure:"appKey"` diff --git a/pkg/common/storage/controller/group.go b/pkg/common/storage/controller/group.go index cb64bc73a..71bf6ead2 100644 --- a/pkg/common/storage/controller/group.go +++ b/pkg/common/storage/controller/group.go @@ -243,17 +243,10 @@ func (g *groupDatabase) UpdateGroup(ctx context.Context, groupID string, data ma if err := g.groupDB.UpdateMap(ctx, groupID, data); err != nil { return err } - userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) - if err != nil { + if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, []string{""}, model.VersionStateUpdate); err != nil { return err } - for _, userID := range userIDs { - if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, userID, []string{groupID}, model.VersionStateUpdate); err != nil { - return err - } - - } - return g.cache.CloneGroupCache().DelGroupsInfo(groupID).DelMaxJoinGroupVersion(userIDs...).ChainExecDel(ctx) + return g.cache.CloneGroupCache().DelGroupsInfo(groupID).DelMaxGroupMemberVersion(groupID).ChainExecDel(ctx) }) } @@ -263,11 +256,11 @@ func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, delete if err := g.groupDB.UpdateStatus(ctx, groupID, constant.GroupStatusDismissed); err != nil { return err } - userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) - if err != nil { - return err - } if deleteMember { + userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) + if err != nil { + return err + } if err := g.groupMemberDB.Delete(ctx, groupID, nil); err != nil { return err } @@ -277,13 +270,18 @@ func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, delete DelGroupMembersHash(groupID). DelGroupAllRoleLevel(groupID). DelGroupMembersInfo(groupID, userIDs...). - DelMaxGroupMemberVersion(groupID) - } - c = c.DelMaxJoinGroupVersion(userIDs...) - if len(userIDs) > 0 { - if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, groupID, userIDs, model.VersionStateDelete); err != nil { + DelMaxGroupMemberVersion(groupID). + DelMaxJoinGroupVersion(userIDs...) + for _, userID := range userIDs { + if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, userID, []string{groupID}, model.VersionStateDelete); err != nil { + return err + } + } + } else { + if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, []string{""}, model.VersionStateUpdate); err != nil { return err } + c = c.DelMaxGroupMemberVersion(groupID) } return c.DelGroupsInfo(groupID).ChainExecDel(ctx) }) @@ -411,12 +409,16 @@ func (g *groupDatabase) TransferGroupOwner(ctx context.Context, groupID string, return c.DelGroupMembersInfo(groupID, oldOwnerUserID, newOwnerUserID). DelGroupAllRoleLevel(groupID). DelGroupMembersHash(groupID). - DelJoinedGroupID(oldOwnerUserID, newOwnerUserID). + DelMaxGroupMemberVersion(groupID). + DelGroupMemberIDs(groupID). ChainExecDel(ctx) }) } func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, userID string, data map[string]any) error { + if len(data) == 0 { + return nil + } return g.ctxTx.Transaction(ctx, func(ctx context.Context) error { if err := g.groupMemberDB.Update(ctx, groupID, userID, data); err != nil { return err @@ -424,9 +426,9 @@ func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, u c := g.cache.CloneGroupCache() c = c.DelGroupMembersInfo(groupID, userID) if g.groupMemberDB.IsUpdateRoleLevel(data) { - c = c.DelGroupAllRoleLevel(groupID) + c = c.DelGroupAllRoleLevel(groupID).DelGroupMemberIDs(groupID) } - c = c.DelMaxGroupMemberVersion(groupID).DelMaxJoinGroupVersion(userID) + c = c.DelMaxGroupMemberVersion(groupID) return c.ChainExecDel(ctx) }) } @@ -439,9 +441,9 @@ func (g *groupDatabase) UpdateGroupMembers(ctx context.Context, data []*common.B return err } if g.groupMemberDB.IsUpdateRoleLevel(item.Map) { - c = c.DelGroupAllRoleLevel(item.GroupID) + c = c.DelGroupAllRoleLevel(item.GroupID).DelGroupMemberIDs(item.GroupID) } - c = c.DelGroupMembersInfo(item.GroupID, item.UserID).DelGroupMembersHash(item.GroupID) + c = c.DelGroupMembersInfo(item.GroupID, item.UserID).DelMaxGroupMemberVersion(item.GroupID).DelGroupMembersHash(item.GroupID) } return c.ChainExecDel(ctx) }) @@ -501,14 +503,6 @@ func (g *groupDatabase) FindJoinIncrVersion(ctx context.Context, userID string, return g.groupMemberDB.FindJoinIncrVersion(ctx, userID, version, limit) } -//func (g *groupDatabase) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) { -// return g.cache.FindSortGroupMemberUserIDs(ctx, groupID) -//} -// -//func (g *groupDatabase) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) { -// return g.cache.FindSortJoinGroupIDs(ctx, userID) -//} - func (g *groupDatabase) FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error) { return g.cache.FindMaxGroupMemberVersion(ctx, groupID) } diff --git a/pkg/common/storage/database/group_member.go b/pkg/common/storage/database/group_member.go index 0051d694f..c49319649 100644 --- a/pkg/common/storage/database/group_member.go +++ b/pkg/common/storage/database/group_member.go @@ -35,6 +35,7 @@ type GroupMember interface { FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) IsUpdateRoleLevel(data map[string]any) bool JoinGroupIncrVersion(ctx context.Context, userID string, groupIDs []string, state int32) error + MemberGroupIncrVersion(ctx context.Context, groupID string, userIDs []string, state int32) error FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error) } diff --git a/pkg/common/storage/database/mgo/friend.go b/pkg/common/storage/database/mgo/friend.go index 18d80d47d..4c3c4b97b 100644 --- a/pkg/common/storage/database/mgo/friend.go +++ b/pkg/common/storage/database/mgo/friend.go @@ -18,6 +18,8 @@ import ( "context" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" + "go.mongodb.org/mongo-driver/bson/primitive" + "time" "github.com/openimsdk/tools/db/mongoutil" "github.com/openimsdk/tools/db/pagination" @@ -53,11 +55,19 @@ func NewFriendMongo(db *mongo.Database) (database.Friend, error) { } func (f *FriendMgo) friendSort() any { - return bson.D{{"is_pinned", -1}, {"create_time", 1}} + return bson.D{{"is_pinned", -1}, {"_id", 1}} } // Create inserts multiple friend records. func (f *FriendMgo) Create(ctx context.Context, friends []*model.Friend) error { + for i, friend := range friends { + if friend.ID.IsZero() { + friends[i].ID = primitive.NewObjectID() + } + if friend.CreateTime.IsZero() { + friends[i].CreateTime = time.Now() + } + } return mongoutil.IncrVersion(func() error { return mongoutil.InsertMany(ctx, f.coll, friends) }, func() error { @@ -108,13 +118,43 @@ func (f *FriendMgo) UpdateRemark(ctx context.Context, ownerUserID, friendUserID, return f.UpdateByMap(ctx, ownerUserID, friendUserID, map[string]any{"remark": remark}) } +func (f *FriendMgo) fillTime(friends ...*model.Friend) { + for i, friend := range friends { + if friend.CreateTime.IsZero() { + friends[i].CreateTime = friend.ID.Timestamp() + } + } +} + +func (f *FriendMgo) findOne(ctx context.Context, filter any) (*model.Friend, error) { + friend, err := mongoutil.FindOne[*model.Friend](ctx, f.coll, filter) + if err != nil { + return nil, err + } + f.fillTime(friend) + return friend, nil +} + +func (f *FriendMgo) find(ctx context.Context, filter any) ([]*model.Friend, error) { + friends, err := mongoutil.Find[*model.Friend](ctx, f.coll, filter) + if err != nil { + return nil, err + } + f.fillTime(friends...) + return friends, nil +} + +func (f *FriendMgo) findPage(ctx context.Context, filter any, pagination pagination.Pagination, opts ...*options.FindOptions) (int64, []*model.Friend, error) { + return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination, opts...) +} + // Take retrieves a single friend document. Returns an error if not found. func (f *FriendMgo) Take(ctx context.Context, ownerUserID, friendUserID string) (*model.Friend, error) { filter := bson.M{ "owner_user_id": ownerUserID, "friend_user_id": friendUserID, } - return mongoutil.FindOne[*model.Friend](ctx, f.coll, filter) + return f.findOne(ctx, filter) } // FindUserState finds the friendship status between two users. @@ -125,7 +165,7 @@ func (f *FriendMgo) FindUserState(ctx context.Context, userID1, userID2 string) {"owner_user_id": userID2, "friend_user_id": userID1}, }, } - return mongoutil.Find[*model.Friend](ctx, f.coll, filter) + return f.find(ctx, filter) } // FindFriends retrieves a list of friends for a given owner. Missing friends do not cause an error. @@ -134,7 +174,7 @@ func (f *FriendMgo) FindFriends(ctx context.Context, ownerUserID string, friendU "owner_user_id": ownerUserID, "friend_user_id": bson.M{"$in": friendUserIDs}, } - return mongoutil.Find[*model.Friend](ctx, f.coll, filter) + return f.find(ctx, filter) } // FindReversalFriends finds users who have added the specified user as a friend. @@ -143,14 +183,14 @@ func (f *FriendMgo) FindReversalFriends(ctx context.Context, friendUserID string "owner_user_id": bson.M{"$in": ownerUserIDs}, "friend_user_id": friendUserID, } - return mongoutil.Find[*model.Friend](ctx, f.coll, filter) + return f.find(ctx, filter) } // FindOwnerFriends retrieves a paginated list of friends for a given owner. func (f *FriendMgo) FindOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (int64, []*model.Friend, error) { filter := bson.M{"owner_user_id": ownerUserID} opt := options.Find().SetSort(f.friendSort()) - return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination, opt) + return f.findPage(ctx, filter, pagination, opt) } func (f *FriendMgo) FindOwnerFriendUserIds(ctx context.Context, ownerUserID string, limit int) ([]string, error) { @@ -162,7 +202,8 @@ func (f *FriendMgo) FindOwnerFriendUserIds(ctx context.Context, ownerUserID stri // FindInWhoseFriends finds users who have added the specified user as a friend, with pagination. func (f *FriendMgo) FindInWhoseFriends(ctx context.Context, friendUserID string, pagination pagination.Pagination) (int64, []*model.Friend, error) { filter := bson.M{"friend_user_id": friendUserID} - return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination) + opt := options.Find().SetSort(f.friendSort()) + return f.findPage(ctx, filter, pagination, opt) } // FindFriendUserIDs retrieves a list of friend user IDs for a given owner. @@ -203,18 +244,3 @@ func (f *FriendMgo) FindFriendUserID(ctx context.Context, friendUserID string) ( } return mongoutil.Find[string](ctx, f.coll, filter, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1}).SetSort(f.friendSort())) } - -//func (f *FriendMgo) SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*model.Friend, error) { -// filter := bson.M{ -// "owner_user_id": ownerUserID, -// } -// if keyword != "" { -// filter["$or"] = []bson.M{ -// {"remark": bson.M{"$regex": keyword, "$options": "i"}}, -// {"nickname": bson.M{"$regex": keyword, "$options": "i"}}, -// {"friend_user_id": bson.M{"$regex": keyword, "$options": "i"}}, -// } -// } -// opt := options.Find().SetSort(f.friendSort()) -// return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination, opt) -//} diff --git a/pkg/common/storage/database/mgo/group_member.go b/pkg/common/storage/database/mgo/group_member.go index a9ac262ab..99a5a3ed2 100644 --- a/pkg/common/storage/database/mgo/group_member.go +++ b/pkg/common/storage/database/mgo/group_member.go @@ -96,13 +96,19 @@ func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []s } return mongoutil.IncrVersion(func() error { return mongoutil.DeleteMany(ctx, g.coll, filter) - }, func() error { - return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete) }, func() error { if len(userIDs) == 0 { - return nil + return g.member.Delete(ctx, groupID) + } else { + return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete) + } + }, func() error { + for _, userID := range userIDs { + if err := g.join.IncrVersion(ctx, userID, []string{groupID}, model.VersionStateDelete); err != nil { + return err + } } - return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete) + return nil }) } @@ -111,8 +117,6 @@ func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, us return g.Update(ctx, groupID, userID, bson.M{"role_level": roleLevel}) }, func() error { return g.member.IncrVersion(ctx, groupID, []string{userID}, model.VersionStateUpdate) - }, func() error { - return g.join.IncrVersion(ctx, groupID, []string{userID}, model.VersionStateUpdate) }) } @@ -178,6 +182,10 @@ func (g *GroupMemberMgo) JoinGroupIncrVersion(ctx context.Context, userID string return g.join.IncrVersion(ctx, userID, groupIDs, state) } +func (g *GroupMemberMgo) MemberGroupIncrVersion(ctx context.Context, groupID string, userIDs []string, state int32) error { + return g.member.IncrVersion(ctx, groupID, userIDs, state) +} + func (g *GroupMemberMgo) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) { return g.member.FindChangeLog(ctx, groupID, version, limit) } diff --git a/pkg/common/storage/database/mgo/version_log.go b/pkg/common/storage/database/mgo/version_log.go index b53057fe4..1d70f96f5 100644 --- a/pkg/common/storage/database/mgo/version_log.go +++ b/pkg/common/storage/database/mgo/version_log.go @@ -256,3 +256,7 @@ func (l *VersionLogMgo) DeleteAfterUnchangedLog(ctx context.Context, deadline ti }, }) } + +func (l *VersionLogMgo) Delete(ctx context.Context, dId string) error { + return mongoutil.DeleteOne(ctx, l.coll, bson.M{"d_id": dId}) +} diff --git a/pkg/common/storage/database/version_log.go b/pkg/common/storage/database/version_log.go index c9dc09540..9d7bcc172 100644 --- a/pkg/common/storage/database/version_log.go +++ b/pkg/common/storage/database/version_log.go @@ -15,4 +15,5 @@ type VersionLog interface { IncrVersion(ctx context.Context, dId string, eIds []string, state int32) error FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error) DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error + Delete(ctx context.Context, dId string) error } diff --git a/pkg/common/storage/model/friend.go b/pkg/common/storage/model/friend.go index 60a40d9c2..abcca2f2b 100644 --- a/pkg/common/storage/model/friend.go +++ b/pkg/common/storage/model/friend.go @@ -15,17 +15,19 @@ package model import ( + "go.mongodb.org/mongo-driver/bson/primitive" "time" ) // Friend represents the data structure for a friend relationship in MongoDB. type Friend struct { - OwnerUserID string `bson:"owner_user_id"` - FriendUserID string `bson:"friend_user_id"` - Remark string `bson:"remark"` - CreateTime time.Time `bson:"create_time"` - AddSource int32 `bson:"add_source"` - OperatorUserID string `bson:"operator_user_id"` - Ex string `bson:"ex"` - IsPinned bool `bson:"is_pinned"` + ID primitive.ObjectID `bson:"_id"` + OwnerUserID string `bson:"owner_user_id"` + FriendUserID string `bson:"friend_user_id"` + Remark string `bson:"remark"` + CreateTime time.Time `bson:"create_time"` + AddSource int32 `bson:"add_source"` + OperatorUserID string `bson:"operator_user_id"` + Ex string `bson:"ex"` + IsPinned bool `bson:"is_pinned"` } diff --git a/pkg/util/hashutil/id.go b/pkg/util/hashutil/id.go new file mode 100644 index 000000000..52e7f4c6f --- /dev/null +++ b/pkg/util/hashutil/id.go @@ -0,0 +1,16 @@ +package hashutil + +import ( + "crypto/md5" + "encoding/binary" + "encoding/json" +) + +func IdHash(ids []string) uint64 { + if len(ids) == 0 { + return 0 + } + data, _ := json.Marshal(ids) + sum := md5.Sum(data) + return binary.BigEndian.Uint64(sum[:]) +}