msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/list' into seq38

Browse Source
pull/2393/head
withchao 1 year ago
parent cce382d0c3 f613c6dfc9
commit 7182298d2f
26 changed files with 329 additions and 155 deletions
Show Stats Download Patch File Download Diff File

4
config/openim-push.yml
Unescape View File

@ -23,7 +23,9 @@ geTui:
channelID: '' channelID: ''
channelName: '' channelName: ''
fcm: 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: jpns:
appKey: '' appKey: ''
masterSecret: '' masterSecret: ''

2
config/redis.yml
Unescape View File

@ -1,4 +1,4 @@
address: [ 172.16.8.48:16379 ] address: [ localhost:16379 ]
username: '' username: ''
password: openIM123 password: openIM123
clusterMode: false clusterMode: false

6
go.mod
Unescape View File

@ -13,7 +13,7 @@ require (
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect
github.com/mitchellh/mapstructure v1.5.0 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/openimsdk/tools v0.0.49-alpha.25
github.com/pkg/errors v0.9.1 // indirect github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0 github.com/prometheus/client_golang v1.18.0
@ -177,4 +177,6 @@ require (
gopkg.in/ini.v1 v1.67.0 // indirect 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
//)

4
go.sum
Unescape View File

@ -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/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 h1:xLDe/moqgWpRoptHzI4packAWzs4C16b+sVY+txNJp0=
github.com/openimsdk/gomake v0.0.13/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= 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.15 h1:iiVROC2nXTc9qPOSfE7NZckQAVCZiDxBm9yY36ULdyw=
github.com/openimsdk/protocol v0.0.69-alpha.9/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= 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 h1:OpRPwDZ2xWX7Zj5kyfZhryu/NfZTrsRVr2GFwu1HQHI=
github.com/openimsdk/tools v0.0.49-alpha.25/go.mod h1:rwsFI1G/nBHNfiNapbven41akRDPBbH4df0Cgy6xueU= 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= github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

4
internal/api/friend.go
Unescape View File

@ -102,3 +102,7 @@ func (o *FriendApi) GetIncrementalFriends(c *gin.Context) {
func (o *FriendApi) GetIncrementalBlacks(c *gin.Context) { func (o *FriendApi) GetIncrementalBlacks(c *gin.Context) {
a2r.Call(relation.FriendClient.GetIncrementalBlacks, o.Client, c) a2r.Call(relation.FriendClient.GetIncrementalBlacks, o.Client, c)
} }
func (o *FriendApi) GetFullFriendUserIDs(c *gin.Context) {
a2r.Call(relation.FriendClient.GetFullFriendUserIDs, o.Client, c)
}

8
internal/api/group.go
Unescape View File

@ -187,10 +187,10 @@ func (o *GroupApi) GetIncrementalGroupMemberBatch(c *gin.Context) {
apiresp.GinSuccess(c, resp) apiresp.GinSuccess(c, resp)
} }
func (o *GroupApi) GetIncrementalGroupMemberUserIDs(c *gin.Context) { func (o *GroupApi) GetFullGroupMemberUserIDs(c *gin.Context) {
a2r.Call(group.GroupClient.GetIncrementalGroupMemberUserIDs, o.Client, c) a2r.Call(group.GroupClient.GetFullGroupMemberUserIDs, o.Client, c)
} }
func (o *GroupApi) GetIncrementalJoinGroupIDs(c *gin.Context) { func (o *GroupApi) GetFullJoinGroupIDs(c *gin.Context) {
a2r.Call(group.GroupClient.GetIncrementalJoinGroupIDs, o.Client, c) a2r.Call(group.GroupClient.GetFullJoinGroupIDs, o.Client, c)
} }

5
internal/api/router.go
Unescape View File

@ -89,6 +89,7 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
friendRouterGroup.POST("/get_specified_friends_info", f.GetSpecifiedFriendsInfo) friendRouterGroup.POST("/get_specified_friends_info", f.GetSpecifiedFriendsInfo)
friendRouterGroup.POST("/update_friends", f.UpdateFriends) friendRouterGroup.POST("/update_friends", f.UpdateFriends)
friendRouterGroup.POST("/get_incremental_friends", f.GetIncrementalFriends) friendRouterGroup.POST("/get_incremental_friends", f.GetIncrementalFriends)
friendRouterGroup.POST("/get_full_friend_user_ids", f.GetFullFriendUserIDs)
} }
g := NewGroupApi(*groupRpc) g := NewGroupApi(*groupRpc)
groupRouterGroup := r.Group("/group") 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_join_group", g.GetIncrementalJoinGroup)
groupRouterGroup.POST("/get_incremental_group_member", g.GetIncrementalGroupMember) groupRouterGroup.POST("/get_incremental_group_member", g.GetIncrementalGroupMember)
groupRouterGroup.POST("/get_incremental_group_member_batch", g.GetIncrementalGroupMemberBatch) groupRouterGroup.POST("/get_incremental_group_member_batch", g.GetIncrementalGroupMemberBatch)
groupRouterGroup.POST("/get_incremental_group_member_user_ids", g.GetIncrementalGroupMemberUserIDs) groupRouterGroup.POST("/get_full_group_member_user_ids", g.GetFullGroupMemberUserIDs)
groupRouterGroup.POST("/get_incremental_join_group_ids", g.GetIncrementalJoinGroupIDs) groupRouterGroup.POST("/get_full_join_group_ids", g.GetFullJoinGroupIDs)
} }
// certificate // certificate
authRouterGroup := r.Group("/auth") authRouterGroup := r.Group("/auth")

46
internal/push/offlinepush/fcm/push.go
Unescape View File

@ -16,8 +16,11 @@ package fcm
import ( import (
"context" "context"
"fmt"
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
"github.com/openimsdk/tools/utils/httputil"
"path/filepath" "path/filepath"
"strings"
firebase "firebase.google.com/go" firebase "firebase.google.com/go"
"firebase.google.com/go/messaging" "firebase.google.com/go/messaging"
@ -40,13 +43,25 @@ type Fcm struct {
// NewClient initializes a new FCM client using the Firebase Admin SDK. // 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. // 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) { func NewClient(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (*Fcm, error) {
projectRoot, err := config.GetProjectRoot() var opt option.ClientOption
if err != nil { switch {
return nil, err 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) fcmApp, err := firebase.NewApp(context.Background(), nil, opt)
if err != nil { if err != nil {
return nil, errs.Wrap(err) return nil, errs.Wrap(err)
@ -56,7 +71,6 @@ func NewClient(pushConf *config.Push, cache cache.ThirdCache) (*Fcm, error) {
if err != nil { if err != nil {
return nil, errs.Wrap(err) return nil, errs.Wrap(err)
} }
return &Fcm{fcmMsgCli: fcmMsgClient, cache: cache}, nil 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.Body = content
notification.Title = title notification.Title = title
var messages []*messaging.Message var messages []*messaging.Message
var sendErrBuilder strings.Builder
var msgErrBuilder strings.Builder
for userID, personTokens := range allTokens { for userID, personTokens := range allTokens {
apns := &messaging.APNSConfig{Payload: &messaging.APNSPayload{Aps: &messaging.Aps{Sound: opts.IOSPushSound}}} apns := &messaging.APNSConfig{Payload: &messaging.APNSPayload{Aps: &messaging.Aps{Sound: opts.IOSPushSound}}}
messageCount := len(messages) 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) response, err := f.fcmMsgCli.SendAll(ctx, messages)
if err != nil { if err != nil {
Fail = Fail + messageCount Fail = Fail + messageCount
// Record push error
sendErrBuilder.WriteString(err.Error())
sendErrBuilder.WriteByte('.')
} else { } else {
Success = Success + response.SuccessCount Success = Success + response.SuccessCount
Fail = Fail + response.FailureCount 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] messages = messages[0:0]
} }
@ -134,5 +162,9 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
Fail = Fail + response.FailureCount 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 return nil
} }

4
internal/push/offlinepush/offlinepusher.go
Unescape View File

@ -36,13 +36,13 @@ type OfflinePusher interface {
Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error 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 var offlinePusher OfflinePusher
switch pushConf.Enable { switch pushConf.Enable {
case geTUI: case geTUI:
offlinePusher = getui.NewClient(pushConf, cache) offlinePusher = getui.NewClient(pushConf, cache)
case firebase: case firebase:
return fcm.NewClient(pushConf, cache) return fcm.NewClient(pushConf, cache, fcmConfigPath)
case jPush: case jPush:
offlinePusher = jpush.NewClient(pushConf) offlinePusher = jpush.NewClient(pushConf)
default: default:

3
internal/push/push.go
Unescape View File

@ -29,6 +29,7 @@ type Config struct {
WebhooksConfig config.Webhooks WebhooksConfig config.Webhooks
LocalCacheConfig config.LocalCache LocalCacheConfig config.LocalCache
Discovery config.Discovery Discovery config.Discovery
FcmConfigPath string
} }
func (p pushServer) PushMsg(ctx context.Context, req *pbpush.PushMsgReq) (*pbpush.PushMsgResp, error) { 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 return err
} }
cacheModel := redis.NewThirdCache(rdb) cacheModel := redis.NewThirdCache(rdb)
offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel) offlinePusher, err := offlinepush.NewOfflinePusher(&config.RpcConfig, cacheModel, config.FcmConfigPath)
if err != nil { if err != nil {
return err return err
} }

24
internal/rpc/auth/auth.go
Unescape View File

@ -16,6 +16,7 @@ package auth
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
redis2 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" redis2 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
"github.com/openimsdk/tools/db/redisutil" "github.com/openimsdk/tools/db/redisutil"
@ -32,7 +33,6 @@ import (
"github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/discovery"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/tokenverify" "github.com/openimsdk/tools/tokenverify"
"google.golang.org/grpc" "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 { if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil {
return nil, err 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 nil, err
} }
return &pbauth.ForceLogoutResp{}, nil 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) conns, err := s.RegisterCenter.GetConns(ctx, s.config.Share.RpcRegisterName.MessageGateway)
if err != nil { if err != nil {
return err return err
} }
for _, v := range conns { for _, v := range conns {
log.ZDebug(ctx, "forceKickOff", "conn", v.Target()) log.ZDebug(ctx, "forceKickOff", "conn", v.Target())
}
for _, v := range conns {
client := msggateway.NewMsgGatewayClient(v) client := msggateway.NewMsgGatewayClient(v)
kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID} kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID}
_, err := client.KickUserOffline(ctx, kickReq) _, 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) 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 return nil
} }
func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) { func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) {
m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID)) m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID))
if err != nil && err != redis.Nil { if err != nil && err != redis.Nil {

42
internal/rpc/friend/friend.go
Unescape View File

@ -41,7 +41,7 @@ import (
) )
type friendServer struct { type friendServer struct {
friendDatabase controller.FriendDatabase db controller.FriendDatabase
blackDatabase controller.BlackDatabase blackDatabase controller.BlackDatabase
userRpcClient *rpcclient.UserRpcClient userRpcClient *rpcclient.UserRpcClient
notificationSender *FriendNotificationSender 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 // Register Friend server with refactored MongoDB and Redis integrations
relation.RegisterFriendServer(server, &friendServer{ relation.RegisterFriendServer(server, &friendServer{
friendDatabase: controller.NewFriendDatabase( db: controller.NewFriendDatabase(
friendMongoDB, friendMongoDB,
friendRequestMongoDB, friendRequestMongoDB,
redis.NewFriendCacheRedis(rdb, &config.LocalCacheConfig, friendMongoDB, redis.GetRocksCacheOptions()), 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 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 { if err != nil {
return nil, err return nil, err
} }
if in1 && in2 { if in1 && in2 {
return nil, servererrs.ErrRelationshipAlready.WrapMsg("already friends has f") 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 return nil, err
} }
s.notificationSender.FriendApplicationAddNotification(ctx, req) s.notificationSender.FriendApplicationAddNotification(ctx, req)
@ -173,7 +173,7 @@ func (s *friendServer) ImportFriends(ctx context.Context, req *relation.ImportFr
return nil, err 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 return nil, err
} }
for _, userID := range req.FriendUserIDs { 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 { if err := s.webhookBeforeAddFriendAgree(ctx, &s.config.WebhooksConfig.BeforeAddFriendAgree, req); err != nil && err != servererrs.ErrCallbackContinue {
return nil, err return nil, err
} }
err := s.friendDatabase.AgreeFriendRequest(ctx, &friendRequest) err := s.db.AgreeFriendRequest(ctx, &friendRequest)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -213,7 +213,7 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *relation.Res
return resp, nil return resp, nil
} }
if req.HandleResult == constant.FriendResponseRefuse { if req.HandleResult == constant.FriendResponseRefuse {
err := s.friendDatabase.RefuseFriendRequest(ctx, &friendRequest) err := s.db.RefuseFriendRequest(ctx, &friendRequest)
if err != nil { if err != nil {
return nil, err 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 { if err := s.userRpcClient.Access(ctx, req.OwnerUserID); err != nil {
return nil, err 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 { if err != nil {
return nil, err 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 return nil, err
} }
s.notificationSender.FriendDeletedNotification(ctx, req) 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 { if err := s.userRpcClient.Access(ctx, req.OwnerUserID); err != nil {
return nil, err 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 { if err != nil {
return nil, err 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 return nil, err
} }
s.webhookAfterSetFriendRemark(ctx, &s.config.WebhooksConfig.AfterSetFriendRemark, req) 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 { if len(friendUserIDs) == 0 {
return nil, nil return nil, nil
} }
friends, err := s.friendDatabase.FindFriendsWithError(ctx, ownerUserID, friendUserIDs) friends, err := s.db.FindFriendsWithError(ctx, ownerUserID, friendUserIDs)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -292,7 +292,7 @@ func (s *friendServer) getFriend(ctx context.Context, ownerUserID string, friend
func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context, func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context,
req *relation.GetDesignatedFriendsApplyReq, req *relation.GetDesignatedFriendsApplyReq,
) (resp *relation.GetDesignatedFriendsApplyResp, err error) { ) (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 { if err != nil {
return nil, err 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 { if err := s.userRpcClient.Access(ctx, req.UserID); err != nil {
return nil, err 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 { if err != nil {
return nil, err 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 { if err := s.userRpcClient.Access(ctx, req.UserID); err != nil {
return nil, err 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 { if err != nil {
return nil, err return nil, err
} }
@ -342,7 +342,7 @@ func (s *friendServer) GetPaginationFriendsApplyFrom(ctx context.Context, req *r
// ok. // ok.
func (s *friendServer) IsFriend(ctx context.Context, req *relation.IsFriendReq) (resp *relation.IsFriendResp, err error) { func (s *friendServer) IsFriend(ctx context.Context, req *relation.IsFriendReq) (resp *relation.IsFriendResp, err error) {
resp = &relation.IsFriendResp{} 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 { if err != nil {
return nil, err 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 { if err := s.userRpcClient.Access(ctx, req.UserID); err != nil {
return nil, err 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 { if err != nil {
return nil, err return nil, err
} }
@ -371,7 +371,7 @@ func (s *friendServer) GetFriendIDs(ctx context.Context, req *relation.GetFriend
return nil, err return nil, err
} }
resp = &relation.GetFriendIDsResp{} resp = &relation.GetFriendIDsResp{}
resp.FriendIDs, err = s.friendDatabase.FindFriendUserIDs(ctx, req.UserID) resp.FriendIDs, err = s.db.FindFriendUserIDs(ctx, req.UserID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -389,7 +389,7 @@ func (s *friendServer) GetSpecifiedFriendsInfo(ctx context.Context, req *relatio
if err != nil { if err != nil {
return nil, err 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 { if err != nil {
return nil, err return nil, err
} }
@ -453,7 +453,7 @@ func (s *friendServer) UpdateFriends(
return nil, errs.ErrArgs.WrapMsg("friendIDList repeated") 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 { if err != nil {
return nil, err return nil, err
} }
@ -469,7 +469,7 @@ func (s *friendServer) UpdateFriends(
if req.Ex != nil { if req.Ex != nil {
val["ex"] = req.Ex.Value 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 return nil, err
} }

26
internal/rpc/friend/sync.go
Unescape View File

@ -2,6 +2,7 @@ package friend
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" "github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
@ -10,6 +11,27 @@ import (
"github.com/openimsdk/protocol/relation" "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) { 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 { if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
return nil, err return nil, err
@ -19,8 +41,8 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.
VersionKey: req.UserID, VersionKey: req.UserID,
VersionID: req.VersionID, VersionID: req.VersionID,
VersionNumber: req.Version, VersionNumber: req.Version,
Version: s.friendDatabase.FindFriendIncrVersion, Version: s.db.FindFriendIncrVersion,
CacheMaxVersion: s.friendDatabase.FindMaxFriendVersionCache, CacheMaxVersion: s.db.FindMaxFriendVersionCache,
Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) { Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) {
return s.getFriend(ctx, req.UserID, ids) return s.getFriend(ctx, req.UserID, ids)
}, },

33
internal/rpc/group/notification.go
Unescape View File

@ -289,10 +289,12 @@ func (g *GroupNotificationSender) fillOpUser(ctx context.Context, opUser **sdkws
return nil return nil
} }
func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint64, collName string, id string) { func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string) {
for _, coll := range versionctx.GetVersionLog(ctx).Get() { versions := versionctx.GetVersionLog(ctx).Get()
for _, coll := range versions {
if coll.Name == collName && coll.Doc.DID == id { if coll.Name == collName && coll.Doc.DID == id {
*version = uint64(coll.Doc.Version) *version = uint64(coll.Doc.Version)
*versionID = coll.Doc.ID.Hex()
return return
} }
} }
@ -308,7 +310,7 @@ func (g *GroupNotificationSender) GroupCreatedNotification(ctx context.Context,
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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()) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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()) 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 return
} }
tips := &sdkws.MemberQuitTips{Group: group, QuitUser: member} 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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} tips := &sdkws.MemberInvitedTips{Group: group, InvitedUserList: users}
err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID) 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) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberInvitedNotification, tips)
} }
@ -540,7 +545,7 @@ func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, g
return return
} }
tips := &sdkws.MemberEnterTips{Group: group, EntrantUser: user} 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return return
} }
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, groupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMutedNotification, tips) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return return
} }
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, groupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupCancelMutedNotification, tips) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) 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 { if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return 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) g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToOrdinaryUserNotification, tips)
} }

77
internal/rpc/group/sync.go
Unescape View File

@ -2,26 +2,18 @@ package group
import ( import (
"context" "context"
"crypto/md5"
"encoding/binary"
"encoding/json"
"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" "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/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/common/storage/model"
"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
"github.com/openimsdk/protocol/constant"
pbgroup "github.com/openimsdk/protocol/group" pbgroup "github.com/openimsdk/protocol/group"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"slices"
) )
func (s *groupServer) idHash(ids []string) uint64 { func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) {
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) {
vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID) vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID)
if err != nil { if err != nil {
return nil, err return nil, err
@ -30,11 +22,11 @@ func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req
if err != nil { if err != nil {
return nil, err return nil, err
} }
idHash := s.idHash(userIDs) idHash := hashutil.IdHash(userIDs)
if req.IdHash == idHash { if req.IdHash == idHash {
userIDs = nil userIDs = nil
} }
return &pbgroup.GetIncrementalGroupMemberUserIDsResp{ return &pbgroup.GetFullGroupMemberUserIDsResp{
Version: idHash, Version: idHash,
VersionID: vl.ID.Hex(), VersionID: vl.ID.Hex(),
Equal: req.IdHash == idHash, Equal: req.IdHash == idHash,
@ -42,7 +34,7 @@ func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req
}, nil }, 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) vl, err := s.db.FindMaxJoinGroupVersionCache(ctx, req.UserID)
if err != nil { if err != nil {
return nil, err return nil, err
@ -51,11 +43,11 @@ func (s *groupServer) GetIncrementalJoinGroupIDs(ctx context.Context, req *pbgro
if err != nil { if err != nil {
return nil, err return nil, err
} }
idHash := s.idHash(groupIDs) idHash := hashutil.IdHash(groupIDs)
if req.IdHash == idHash { if req.IdHash == idHash {
groupIDs = nil groupIDs = nil
} }
return &pbgroup.GetIncrementalJoinGroupIDsResp{ return &pbgroup.GetFullJoinGroupIDsResp{
Version: idHash, Version: idHash,
VersionID: vl.ID.Hex(), VersionID: vl.ID.Hex(),
Equal: req.IdHash == idHash, 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) { 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]{ opt := incrversion.Option[*sdkws.GroupMemberFullInfo, pbgroup.GetIncrementalGroupMemberResp]{
Ctx: ctx, Ctx: ctx,
VersionKey: req.GroupID, VersionKey: req.GroupID,
VersionID: req.VersionID, VersionID: req.VersionID,
VersionNumber: req.Version, VersionNumber: req.Version,
Version: s.db.FindMemberIncrVersion, 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, CacheMaxVersion: s.db.FindMaxGroupMemberVersionCache,
Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) { Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) {
return s.getGroupMembersInfo(ctx, req.GroupID, ids) 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) { func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupReq) (*pbgroup.GetIncrementalJoinGroupResp, error) {

1
pkg/common/cmd/push.go
Unescape View File

@ -47,6 +47,7 @@ func NewPushRpcCmd() *PushRpcCmd {
ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap)) ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap))
ret.ctx = context.WithValue(context.Background(), "version", config.Version) ret.ctx = context.WithValue(context.Background(), "version", config.Version)
ret.Command.RunE = func(cmd *cobra.Command, args []string) error { ret.Command.RunE = func(cmd *cobra.Command, args []string) error {
ret.pushConfig.FcmConfigPath = ret.ConfigPath()
return ret.runE() return ret.runE()
} }
return ret return ret

6
pkg/common/cmd/root.go
Unescape View File

@ -31,6 +31,11 @@ type RootCmd struct {
prometheusPort int prometheusPort int
log config.Log log config.Log
index int index int
configPath string
}
func (r *RootCmd) ConfigPath() string {
return r.configPath
} }
func (r *RootCmd) Index() int { func (r *RootCmd) Index() int {
@ -153,6 +158,7 @@ func (r *RootCmd) getFlag(cmd *cobra.Command) (string, int, error) {
if err != nil { if err != nil {
return "", 0, errs.Wrap(err) return "", 0, errs.Wrap(err)
} }
r.configPath = configDirectory
index, err := cmd.Flags().GetInt(FlagTransferIndex) index, err := cmd.Flags().GetInt(FlagTransferIndex)
if err != nil { if err != nil {
return "", 0, errs.Wrap(err) return "", 0, errs.Wrap(err)

3
pkg/common/config/config.go
Unescape View File

@ -202,7 +202,8 @@ type Push struct {
ChannelName string `mapstructure:"channelName"` ChannelName string `mapstructure:"channelName"`
} `mapstructure:"geTui"` } `mapstructure:"geTui"`
FCM struct { FCM struct {
ServiceAccount string `mapstructure:"serviceAccount"` FilePath string `mapstructure:"filePath"`
AuthURL string `mapstructure:"authURL"`
} `mapstructure:"fcm"` } `mapstructure:"fcm"`
JPNS struct { JPNS struct {
AppKey string `mapstructure:"appKey"` AppKey string `mapstructure:"appKey"`

56
pkg/common/storage/controller/group.go
Unescape View File

@ -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 { if err := g.groupDB.UpdateMap(ctx, groupID, data); err != nil {
return err return err
} }
userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, []string{""}, model.VersionStateUpdate); err != nil {
if err != nil {
return err return err
} }
for _, userID := range userIDs { return g.cache.CloneGroupCache().DelGroupsInfo(groupID).DelMaxGroupMemberVersion(groupID).ChainExecDel(ctx)
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)
}) })
} }
@ -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 { if err := g.groupDB.UpdateStatus(ctx, groupID, constant.GroupStatusDismissed); err != nil {
return err return err
} }
userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID)
if err != nil {
return err
}
if deleteMember { if deleteMember {
userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID)
if err != nil {
return err
}
if err := g.groupMemberDB.Delete(ctx, groupID, nil); err != nil { if err := g.groupMemberDB.Delete(ctx, groupID, nil); err != nil {
return err return err
} }
@ -277,13 +270,18 @@ func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, delete
DelGroupMembersHash(groupID). DelGroupMembersHash(groupID).
DelGroupAllRoleLevel(groupID). DelGroupAllRoleLevel(groupID).
DelGroupMembersInfo(groupID, userIDs...). DelGroupMembersInfo(groupID, userIDs...).
DelMaxGroupMemberVersion(groupID) DelMaxGroupMemberVersion(groupID).
} DelMaxJoinGroupVersion(userIDs...)
c = c.DelMaxJoinGroupVersion(userIDs...) for _, userID := range userIDs {
if len(userIDs) > 0 { if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, userID, []string{groupID}, model.VersionStateDelete); err != nil {
if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, groupID, userIDs, model.VersionStateDelete); err != nil { return err
}
}
} else {
if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, []string{""}, model.VersionStateUpdate); err != nil {
return err return err
} }
c = c.DelMaxGroupMemberVersion(groupID)
} }
return c.DelGroupsInfo(groupID).ChainExecDel(ctx) 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). return c.DelGroupMembersInfo(groupID, oldOwnerUserID, newOwnerUserID).
DelGroupAllRoleLevel(groupID). DelGroupAllRoleLevel(groupID).
DelGroupMembersHash(groupID). DelGroupMembersHash(groupID).
DelJoinedGroupID(oldOwnerUserID, newOwnerUserID). DelMaxGroupMemberVersion(groupID).
DelGroupMemberIDs(groupID).
ChainExecDel(ctx) ChainExecDel(ctx)
}) })
} }
func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, userID string, data map[string]any) error { 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 { return g.ctxTx.Transaction(ctx, func(ctx context.Context) error {
if err := g.groupMemberDB.Update(ctx, groupID, userID, data); err != nil { if err := g.groupMemberDB.Update(ctx, groupID, userID, data); err != nil {
return err return err
@ -424,9 +426,9 @@ func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, u
c := g.cache.CloneGroupCache() c := g.cache.CloneGroupCache()
c = c.DelGroupMembersInfo(groupID, userID) c = c.DelGroupMembersInfo(groupID, userID)
if g.groupMemberDB.IsUpdateRoleLevel(data) { 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) return c.ChainExecDel(ctx)
}) })
} }
@ -439,9 +441,9 @@ func (g *groupDatabase) UpdateGroupMembers(ctx context.Context, data []*common.B
return err return err
} }
if g.groupMemberDB.IsUpdateRoleLevel(item.Map) { 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) 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) 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) { func (g *groupDatabase) FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error) {
return g.cache.FindMaxGroupMemberVersion(ctx, groupID) return g.cache.FindMaxGroupMemberVersion(ctx, groupID)
} }

1
pkg/common/storage/database/group_member.go
Unescape View File

@ -35,6 +35,7 @@ type GroupMember interface {
FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error)
IsUpdateRoleLevel(data map[string]any) bool IsUpdateRoleLevel(data map[string]any) bool
JoinGroupIncrVersion(ctx context.Context, userID string, groupIDs []string, state int32) error 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) 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) FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error)
} }

70
pkg/common/storage/database/mgo/friend.go
Unescape View File

@ -18,6 +18,8 @@ import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" "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/model"
"go.mongodb.org/mongo-driver/bson/primitive"
"time"
"github.com/openimsdk/tools/db/mongoutil" "github.com/openimsdk/tools/db/mongoutil"
"github.com/openimsdk/tools/db/pagination" "github.com/openimsdk/tools/db/pagination"
@ -53,11 +55,19 @@ func NewFriendMongo(db *mongo.Database) (database.Friend, error) {
} }
func (f *FriendMgo) friendSort() any { 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. // Create inserts multiple friend records.
func (f *FriendMgo) Create(ctx context.Context, friends []*model.Friend) error { 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.IncrVersion(func() error {
return mongoutil.InsertMany(ctx, f.coll, friends) return mongoutil.InsertMany(ctx, f.coll, friends)
}, func() error { }, 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}) 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. // 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) { func (f *FriendMgo) Take(ctx context.Context, ownerUserID, friendUserID string) (*model.Friend, error) {
filter := bson.M{ filter := bson.M{
"owner_user_id": ownerUserID, "owner_user_id": ownerUserID,
"friend_user_id": friendUserID, "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. // 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}, {"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. // 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, "owner_user_id": ownerUserID,
"friend_user_id": bson.M{"$in": friendUserIDs}, "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. // 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}, "owner_user_id": bson.M{"$in": ownerUserIDs},
"friend_user_id": friendUserID, "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. // 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) { func (f *FriendMgo) FindOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (int64, []*model.Friend, error) {
filter := bson.M{"owner_user_id": ownerUserID} filter := bson.M{"owner_user_id": ownerUserID}
opt := options.Find().SetSort(f.friendSort()) 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) { 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. // 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) { func (f *FriendMgo) FindInWhoseFriends(ctx context.Context, friendUserID string, pagination pagination.Pagination) (int64, []*model.Friend, error) {
filter := bson.M{"friend_user_id": friendUserID} 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. // 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())) 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)
//}

20
pkg/common/storage/database/mgo/group_member.go
Unescape View File

@ -96,13 +96,19 @@ func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []s
} }
return mongoutil.IncrVersion(func() error { return mongoutil.IncrVersion(func() error {
return mongoutil.DeleteMany(ctx, g.coll, filter) return mongoutil.DeleteMany(ctx, g.coll, filter)
}, func() error {
return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete)
}, func() error { }, func() error {
if len(userIDs) == 0 { 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}) return g.Update(ctx, groupID, userID, bson.M{"role_level": roleLevel})
}, func() error { }, func() error {
return g.member.IncrVersion(ctx, groupID, []string{userID}, model.VersionStateUpdate) 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) 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) { func (g *GroupMemberMgo) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) {
return g.member.FindChangeLog(ctx, groupID, version, limit) return g.member.FindChangeLog(ctx, groupID, version, limit)
} }

4
pkg/common/storage/database/mgo/version_log.go
Unescape View File

@ -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})
}

1
pkg/common/storage/database/version_log.go
Unescape View File

@ -15,4 +15,5 @@ type VersionLog interface {
IncrVersion(ctx context.Context, dId string, eIds []string, state int32) error IncrVersion(ctx context.Context, dId string, eIds []string, state int32) error
FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error) FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error)
DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error
Delete(ctx context.Context, dId string) error
} }

18
pkg/common/storage/model/friend.go
Unescape View File

@ -15,17 +15,19 @@
package model package model
import ( import (
"go.mongodb.org/mongo-driver/bson/primitive"
"time" "time"
) )
// Friend represents the data structure for a friend relationship in MongoDB. // Friend represents the data structure for a friend relationship in MongoDB.
type Friend struct { type Friend struct {
OwnerUserID string `bson:"owner_user_id"` ID primitive.ObjectID `bson:"_id"`
FriendUserID string `bson:"friend_user_id"` OwnerUserID string `bson:"owner_user_id"`
Remark string `bson:"remark"` FriendUserID string `bson:"friend_user_id"`
CreateTime time.Time `bson:"create_time"` Remark string `bson:"remark"`
AddSource int32 `bson:"add_source"` CreateTime time.Time `bson:"create_time"`
OperatorUserID string `bson:"operator_user_id"` AddSource int32 `bson:"add_source"`
Ex string `bson:"ex"` OperatorUserID string `bson:"operator_user_id"`
IsPinned bool `bson:"is_pinned"` Ex string `bson:"ex"`
IsPinned bool `bson:"is_pinned"`
} }

16
pkg/util/hashutil/id.go
Unescape View File

@ -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[:])
}
Write Preview
Loading…
Cancel
Save