Xinwei Xiong
2 years ago
committed by
GitHub
commit
76e32ddfcf
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,132 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"github.com/OpenIMSDK/protocol/user"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo"
|
||||
tablerelation "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
tx2 "github.com/openimsdk/open-im-server/v3/pkg/common/db/tx"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"math/rand"
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
)
|
||||
|
||||
var (
|
||||
rdb redis.UniversalClient
|
||||
mgo *unrelation.Mongo
|
||||
ctx context.Context
|
||||
)
|
||||
|
||||
func InitDB() error {
|
||||
addr := "172.16.8.142"
|
||||
pwd := "openIM123"
|
||||
|
||||
config.Config.Redis.Address = []string{net.JoinHostPort(addr, "16379")}
|
||||
config.Config.Redis.Password = pwd
|
||||
config.Config.Mongo.Address = []string{net.JoinHostPort(addr, "37017")}
|
||||
config.Config.Mongo.Database = "openIM_v3"
|
||||
config.Config.Mongo.Username = "root"
|
||||
config.Config.Mongo.Password = pwd
|
||||
config.Config.Mongo.MaxPoolSize = 100
|
||||
var err error
|
||||
rdb, err = cache.NewRedis()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mgo, err = unrelation.NewMongo()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tx, err := tx2.NewAuto(context.Background(), mgo.GetClient())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
config.Config.Object.Enable = "minio"
|
||||
config.Config.Object.ApiURL = "http://" + net.JoinHostPort(addr, "10002")
|
||||
config.Config.Object.Minio.Bucket = "openim"
|
||||
config.Config.Object.Minio.Endpoint = "http://" + net.JoinHostPort(addr, "10005")
|
||||
config.Config.Object.Minio.AccessKeyID = "root"
|
||||
config.Config.Object.Minio.SecretAccessKey = pwd
|
||||
config.Config.Object.Minio.SignEndpoint = config.Config.Object.Minio.Endpoint
|
||||
|
||||
config.Config.Manager.UserID = []string{"openIM123456"}
|
||||
config.Config.Manager.Nickname = []string{"openIM123456"}
|
||||
|
||||
ctx = context.WithValue(context.Background(), constant.OperationID, "debugOperationID")
|
||||
ctx = context.WithValue(context.Background(), constant.OpUserID, config.Config.Manager.UserID[0])
|
||||
|
||||
if err := log.InitFromConfig("", "", 6, true, false, "", 2, 1); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
users := make([]*tablerelation.UserModel, 0)
|
||||
if len(config.Config.Manager.UserID) != len(config.Config.Manager.Nickname) {
|
||||
return errors.New("len(config.Config.Manager.AppManagerUid) != len(config.Config.Manager.Nickname)")
|
||||
}
|
||||
for k, v := range config.Config.Manager.UserID {
|
||||
users = append(users, &tablerelation.UserModel{UserID: v, Nickname: config.Config.Manager.Nickname[k], AppMangerLevel: constant.AppAdmin})
|
||||
}
|
||||
userDB, err := newmgo.NewUserMongo(mgo.GetDatabase())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
//var client registry.SvcDiscoveryRegistry
|
||||
//_= client
|
||||
cache := cache.NewUserCacheRedis(rdb, userDB, cache.GetDefaultOpt())
|
||||
userMongoDB := unrelation.NewUserMongoDriver(mgo.GetDatabase())
|
||||
database := controller.NewUserDatabase(userDB, cache, tx, userMongoDB)
|
||||
//friendRpcClient := rpcclient.NewFriendRpcClient(client)
|
||||
//groupRpcClient := rpcclient.NewGroupRpcClient(client)
|
||||
//msgRpcClient := rpcclient.NewMessageRpcClient(client)
|
||||
|
||||
userSrv = &userServer{
|
||||
UserDatabase: database,
|
||||
//RegisterCenter: client,
|
||||
//friendRpcClient: &friendRpcClient,
|
||||
//groupRpcClient: &groupRpcClient,
|
||||
//friendNotificationSender: notification.NewFriendNotificationSender(&msgRpcClient, notification.WithDBFunc(database.FindWithError)),
|
||||
//userNotificationSender: notification.NewUserNotificationSender(&msgRpcClient, notification.WithUserFunc(database.FindWithError)),
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
if err := InitDB(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
var userSrv *userServer
|
||||
|
||||
func TestName(t *testing.T) {
|
||||
userID := strconv.Itoa(int(rand.Uint32()))
|
||||
res, err := userSrv.UserRegister(ctx, &user.UserRegisterReq{
|
||||
Secret: config.Config.Secret,
|
||||
Users: []*sdkws.UserInfo{
|
||||
{
|
||||
UserID: userID,
|
||||
Nickname: userID,
|
||||
FaceURL: "",
|
||||
Ex: "",
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
t.Log(res)
|
||||
|
||||
}
|
||||
@ -0,0 +1,86 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package convert
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
)
|
||||
|
||||
func TestUsersDB2Pb(t *testing.T) {
|
||||
type args struct {
|
||||
users []*relationtb.UserModel
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
wantResult []*sdkws.UserInfo
|
||||
}{
|
||||
// TODO: Add test cases.
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if gotResult := UsersDB2Pb(tt.args.users); !reflect.DeepEqual(gotResult, tt.wantResult) {
|
||||
t.Errorf("UsersDB2Pb() = %v, want %v", gotResult, tt.wantResult)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUserPb2DB(t *testing.T) {
|
||||
type args struct {
|
||||
user *sdkws.UserInfo
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want *relationtb.UserModel
|
||||
}{
|
||||
// TODO: Add test cases.
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := UserPb2DB(tt.args.user); !reflect.DeepEqual(got, tt.want) {
|
||||
t.Errorf("UserPb2DB() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUserPb2DBMap(t *testing.T) {
|
||||
user := &sdkws.UserInfo{
|
||||
Nickname: "TestUser",
|
||||
FaceURL: "http://openim.io/logo.jpg",
|
||||
Ex: "Extra Data",
|
||||
AppMangerLevel: 1,
|
||||
GlobalRecvMsgOpt: 2,
|
||||
}
|
||||
|
||||
expected := map[string]any{
|
||||
"nickname": "TestUser",
|
||||
"face_url": "http://openim.io/logo.jpg",
|
||||
"ex": "Extra Data",
|
||||
"app_manager_level": int32(1),
|
||||
"global_recv_msg_opt": int32(2),
|
||||
}
|
||||
|
||||
result := UserPb2DBMap(user)
|
||||
if !reflect.DeepEqual(result, expected) {
|
||||
t.Errorf("UserPb2DBMap returned unexpected map. Got %v, want %v", result, expected)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,129 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewConversationMongo(db *mongo.Database) (*ConversationMgo, error) {
|
||||
return &ConversationMgo{
|
||||
coll: db.Collection("conversation"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
type ConversationMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) Create(ctx context.Context, conversations []*relation.ConversationModel) (err error) {
|
||||
return mgotool.InsertMany(ctx, c.coll, conversations)
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) Delete(ctx context.Context, groupIDs []string) (err error) {
|
||||
return mgotool.DeleteMany(ctx, c.coll, bson.M{"group_id": bson.M{"$in": groupIDs}})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) UpdateByMap(ctx context.Context, userIDs []string, conversationID string, args map[string]any) (rows int64, err error) {
|
||||
res, err := mgotool.UpdateMany(ctx, c.coll, bson.M{"owner_user_id": bson.M{"$in": userIDs}, "conversation_id": conversationID}, bson.M{"$set": args})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return res.ModifiedCount, nil
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) Update(ctx context.Context, conversation *relation.ConversationModel) (err error) {
|
||||
return mgotool.UpdateOne(ctx, c.coll, bson.M{"owner_user_id": conversation.OwnerUserID, "conversation_id": conversation.ConversationID}, bson.M{"$set": conversation}, true)
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) Find(ctx context.Context, ownerUserID string, conversationIDs []string) (conversations []*relation.ConversationModel, err error) {
|
||||
return mgotool.Find[*relation.ConversationModel](ctx, c.coll, bson.M{"owner_user_id": ownerUserID, "conversation_id": bson.M{"$in": conversationIDs}})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) FindUserID(ctx context.Context, userIDs []string, conversationIDs []string) ([]string, error) {
|
||||
return mgotool.Find[string](ctx, c.coll, bson.M{"owner_user_id": bson.M{"$in": userIDs}, "conversation_id": bson.M{"$in": conversationIDs}}, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1}))
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) FindUserIDAllConversationID(ctx context.Context, userID string) ([]string, error) {
|
||||
return mgotool.Find[string](ctx, c.coll, bson.M{"owner_user_id": userID}, options.Find().SetProjection(bson.M{"_id": 0, "conversation_id": 1}))
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) Take(ctx context.Context, userID, conversationID string) (conversation *relation.ConversationModel, err error) {
|
||||
return mgotool.FindOne[*relation.ConversationModel](ctx, c.coll, bson.M{"owner_user_id": userID, "conversation_id": conversationID})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) FindConversationID(ctx context.Context, userID string, conversationIDs []string) (existConversationID []string, err error) {
|
||||
return mgotool.Find[string](ctx, c.coll, bson.M{"owner_user_id": userID, "conversation_id": bson.M{"$in": conversationIDs}}, options.Find().SetProjection(bson.M{"_id": 0, "conversation_id": 1}))
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) FindUserIDAllConversations(ctx context.Context, userID string) (conversations []*relation.ConversationModel, err error) {
|
||||
return mgotool.Find[*relation.ConversationModel](ctx, c.coll, bson.M{"owner_user_id": userID})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) FindRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error) {
|
||||
return mgotool.Find[string](ctx, c.coll, bson.M{"group_id": groupID, "recv_msg_opt": constant.ReceiveNotNotifyMessage}, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1}))
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error) {
|
||||
return mgotool.FindOne[int](ctx, c.coll, bson.M{"owner_user_id": ownerUserID, "conversation_id": conversationID}, options.FindOne().SetProjection(bson.M{"recv_msg_opt": 1}))
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) GetAllConversationIDs(ctx context.Context) ([]string, error) {
|
||||
return mgotool.Aggregate[string](ctx, c.coll, []bson.M{
|
||||
{"$group": bson.M{"_id": "$conversation_id"}},
|
||||
{"$project": bson.M{"_id": 0, "conversation_id": "$_id"}},
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) GetAllConversationIDsNumber(ctx context.Context) (int64, error) {
|
||||
counts, err := mgotool.Aggregate[int64](ctx, c.coll, []bson.M{
|
||||
{"$group": bson.M{"_id": "$conversation_id"}},
|
||||
{"$project": bson.M{"_id": 0, "conversation_id": "$_id"}},
|
||||
})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(counts) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
return counts[0], nil
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) PageConversationIDs(ctx context.Context, pagination pagination.Pagination) (conversationIDs []string, err error) {
|
||||
return mgotool.FindPageOnly[string](ctx, c.coll, bson.M{}, pagination, options.Find().SetProjection(bson.M{"conversation_id": 1}))
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relation.ConversationModel, error) {
|
||||
return mgotool.Find[*relation.ConversationModel](ctx, c.coll, bson.M{"conversation_id": bson.M{"$in": conversationIDs}})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relation.ConversationModel, error) {
|
||||
//"is_msg_destruct = 1 && msg_destruct_time != 0 && (UNIX_TIMESTAMP(NOW()) > (msg_destruct_time + UNIX_TIMESTAMP(latest_msg_destruct_time)) || latest_msg_destruct_time is NULL)"
|
||||
return mgotool.Find[*relation.ConversationModel](ctx, c.coll, bson.M{
|
||||
"is_msg_destruct": 1,
|
||||
"msg_destruct_time": bson.M{"$ne": 0},
|
||||
"$or": []bson.M{
|
||||
{
|
||||
"$expr": bson.M{
|
||||
"$gt": []any{
|
||||
time.Now(),
|
||||
bson.M{"$add": []any{"$msg_destruct_time", "$latest_msg_destruct_time"}},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"latest_msg_destruct_time": nil,
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationMgo) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
return mgotool.Find[string](ctx, c.coll, bson.M{"conversation_id": conversationID, "recv_msg_opt": bson.M{"$ne": constant.ReceiveMessage}}, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1}))
|
||||
}
|
||||
@ -0,0 +1,169 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
)
|
||||
|
||||
// FriendMgo implements FriendModelInterface using MongoDB as the storage backend.
|
||||
type FriendMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
// NewFriendMongo creates a new instance of FriendMgo with the provided MongoDB database.
|
||||
func NewFriendMongo(db *mongo.Database) (relation.FriendModelInterface, error) {
|
||||
return &FriendMgo{
|
||||
coll: db.Collection(relation.FriendModelCollectionName),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Create inserts multiple friend records.
|
||||
func (f *FriendMgo) Create(ctx context.Context, friends []*relation.FriendModel) error {
|
||||
return mgotool.InsertMany(ctx, f.coll, friends)
|
||||
}
|
||||
|
||||
// Delete removes specified friends of the owner user.
|
||||
func (f *FriendMgo) Delete(ctx context.Context, ownerUserID string, friendUserIDs []string) error {
|
||||
filter := bson.M{
|
||||
"owner_user_id": ownerUserID,
|
||||
"friend_user_id": bson.M{"$in": friendUserIDs},
|
||||
}
|
||||
_, err := f.coll.DeleteMany(ctx, filter)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateByMap updates specific fields of a friend document using a map.
|
||||
func (f *FriendMgo) UpdateByMap(ctx context.Context, ownerUserID string, friendUserID string, args map[string]interface{}) error {
|
||||
if len(args) == 0 {
|
||||
return nil
|
||||
}
|
||||
filter := bson.M{
|
||||
"owner_user_id": ownerUserID,
|
||||
"friend_user_id": friendUserID,
|
||||
}
|
||||
update := bson.M{"$set": args}
|
||||
err := mgotool.UpdateOne(ctx, f.coll, filter, update, true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Update modifies multiple friend documents.
|
||||
// func (f *FriendMgo) Update(ctx context.Context, friends []*relation.FriendModel) error {
|
||||
// filter := bson.M{
|
||||
// "owner_user_id": ownerUserID,
|
||||
// "friend_user_id": friendUserID,
|
||||
// }
|
||||
// return mgotool.UpdateMany(ctx, f.coll, filter, friends)
|
||||
// }
|
||||
|
||||
// UpdateRemark updates the remark for a specific friend.
|
||||
func (f *FriendMgo) UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) error {
|
||||
args := map[string]interface{}{"remark": remark}
|
||||
return f.UpdateByMap(ctx, ownerUserID, friendUserID, args)
|
||||
}
|
||||
|
||||
// Take retrieves a single friend document. Returns an error if not found.
|
||||
func (f *FriendMgo) Take(ctx context.Context, ownerUserID, friendUserID string) (*relation.FriendModel, error) {
|
||||
filter := bson.M{
|
||||
"owner_user_id": ownerUserID,
|
||||
"friend_user_id": friendUserID,
|
||||
}
|
||||
friend, err := mgotool.FindOne[*relation.FriendModel](ctx, f.coll, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return friend, nil
|
||||
}
|
||||
|
||||
// FindUserState finds the friendship status between two users.
|
||||
func (f *FriendMgo) FindUserState(ctx context.Context, userID1, userID2 string) ([]*relation.FriendModel, error) {
|
||||
filter := bson.M{
|
||||
"$or": []bson.M{
|
||||
{"owner_user_id": userID1, "friend_user_id": userID2},
|
||||
{"owner_user_id": userID2, "friend_user_id": userID1},
|
||||
},
|
||||
}
|
||||
friends, err := mgotool.Find[*relation.FriendModel](ctx, f.coll, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return friends, nil
|
||||
}
|
||||
|
||||
// FindFriends retrieves a list of friends for a given owner. Missing friends do not cause an error.
|
||||
func (f *FriendMgo) FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) ([]*relation.FriendModel, error) {
|
||||
filter := bson.M{
|
||||
"owner_user_id": ownerUserID,
|
||||
"friend_user_id": bson.M{"$in": friendUserIDs},
|
||||
}
|
||||
friends, err := mgotool.Find[*relation.FriendModel](ctx, f.coll, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return friends, nil
|
||||
}
|
||||
|
||||
// FindReversalFriends finds users who have added the specified user as a friend.
|
||||
func (f *FriendMgo) FindReversalFriends(ctx context.Context, friendUserID string, ownerUserIDs []string) ([]*relation.FriendModel, error) {
|
||||
filter := bson.M{
|
||||
"owner_user_id": bson.M{"$in": ownerUserIDs},
|
||||
"friend_user_id": friendUserID,
|
||||
}
|
||||
friends, err := mgotool.Find[*relation.FriendModel](ctx, f.coll, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return friends, nil
|
||||
}
|
||||
|
||||
// FindOwnerFriends retrieves a paginated list of friends for a given owner.
|
||||
func (f *FriendMgo) FindOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination, showNumber int32) ([]*relation.FriendModel, int64, error) {
|
||||
filter := bson.M{"owner_user_id": ownerUserID}
|
||||
count, friends, err := mgotool.FindPage[*relation.FriendModel](ctx, f.coll, filter, pagination)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return friends, count, nil
|
||||
}
|
||||
|
||||
// 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, showNumber int32) ([]*relation.FriendModel, int64, error) {
|
||||
filter := bson.M{"friend_user_id": friendUserID}
|
||||
count, friends, err := mgotool.FindPage[*relation.FriendModel](ctx, f.coll, filter, pagination)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return friends, count, nil
|
||||
}
|
||||
|
||||
// FindFriendUserIDs retrieves a list of friend user IDs for a given owner.
|
||||
func (f *FriendMgo) FindFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) {
|
||||
filter := bson.M{"owner_user_id": ownerUserID}
|
||||
friends := []*relation.FriendModel{}
|
||||
friends, err := mgotool.Find[*relation.FriendModel](ctx, f.coll, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
friendUserIDs := make([]string, len(friends))
|
||||
for i, friend := range friends {
|
||||
friendUserIDs[i] = friend.FriendUserID
|
||||
}
|
||||
return friendUserIDs, nil
|
||||
}
|
||||
|
||||
// NewTx creates a new transaction.
|
||||
func (f *FriendMgo) NewTx(tx any) relation.FriendModelInterface {
|
||||
panic("not implemented")
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
)
|
||||
|
||||
type FriendRequestMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func NewFriendRequestMongo(db *mongo.Database) (relation.FriendRequestModelInterface, error) {
|
||||
return &FriendRequestMgo{
|
||||
coll: db.Collection(relation.FriendRequestModelCollectionName),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (f *FriendRequestMgo) Create(ctx context.Context, friendRequests []*relation.FriendRequestModel) error {
|
||||
return mgotool.InsertMany(ctx, f.coll, friendRequests)
|
||||
}
|
||||
|
||||
func (f *FriendRequestMgo) Delete(ctx context.Context, fromUserID, toUserID string) (err error) {
|
||||
return mgotool.Delete[*relation.FriendRequestModel](ctx, f.coll, bson.M{"from_user_id": fromUserID, "to_user_id": toUserID})
|
||||
}
|
||||
|
||||
func (f *FriendRequestMgo) UpdateByMap(ctx context.Context, formUserID, toUserID string, args map[string]any) (err error) {
|
||||
if len(args) == 0 {
|
||||
return nil
|
||||
}
|
||||
return mgotool.UpdateOne(ctx, f.coll, bson.M{"from_user_id": formUserID, "to_user_id": toUserID}, bson.M{"$set": args}, true)
|
||||
}
|
||||
|
||||
func (f *FriendRequestMgo) Update(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) {
|
||||
return mgotool.UpdateOne(ctx, f.coll, bson.M{"_id": friendRequest.ID}, bson.M{"$set": friendRequest}, true)
|
||||
}
|
||||
|
||||
func (f *FriendRequestMgo) Find(ctx context.Context, fromUserID, toUserID string) (friendRequest *relation.FriendRequestModel, err error) {
|
||||
return mgotool.FindOne[*relation.FriendRequestModel](ctx, f.coll, bson.M{"from_user_id": fromUserID, "to_user_id": toUserID})
|
||||
}
|
||||
|
||||
func (f *FriendRequestMgo) Take(ctx context.Context, fromUserID, toUserID string) (friendRequest *relation.FriendRequestModel, err error) {
|
||||
return f.Find(ctx, fromUserID, toUserID)
|
||||
}
|
||||
|
||||
@ -0,0 +1,60 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewGroupMongo(db *mongo.Database) (relation.GroupModelInterface, error) {
|
||||
return &GroupMgo{
|
||||
coll: db.Collection("group"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
type GroupMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (g *GroupMgo) Create(ctx context.Context, groups []*relation.GroupModel) (err error) {
|
||||
return mgotool.InsertMany(ctx, g.coll, groups)
|
||||
}
|
||||
|
||||
func (g *GroupMgo) UpdateState(ctx context.Context, groupID string, state int32) (err error) {
|
||||
return g.UpdateMap(ctx, groupID, map[string]any{"state": state})
|
||||
}
|
||||
|
||||
func (g *GroupMgo) UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error) {
|
||||
if len(args) == 0 {
|
||||
return nil
|
||||
}
|
||||
return mgotool.UpdateOne(ctx, g.coll, bson.M{"group_id": groupID}, bson.M{"$set": args}, true)
|
||||
}
|
||||
|
||||
func (g *GroupMgo) Find(ctx context.Context, groupIDs []string) (groups []*relation.GroupModel, err error) {
|
||||
return mgotool.Find[*relation.GroupModel](ctx, g.coll, bson.M{"group_id": bson.M{"$in": groupIDs}})
|
||||
}
|
||||
|
||||
func (g *GroupMgo) Take(ctx context.Context, groupID string) (group *relation.GroupModel, err error) {
|
||||
return mgotool.FindOne[*relation.GroupModel](ctx, g.coll, bson.M{"group_id": groupID})
|
||||
}
|
||||
|
||||
func (g *GroupMgo) Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, groups []*relation.GroupModel, err error) {
|
||||
return mgotool.FindPage[*relation.GroupModel](ctx, g.coll, bson.M{"group_name": bson.M{"$regex": keyword}}, pagination)
|
||||
}
|
||||
|
||||
func (g *GroupMgo) CountTotal(ctx context.Context, before *time.Time) (count int64, err error) {
|
||||
if before == nil {
|
||||
return mgotool.Count(ctx, g.coll, bson.M{})
|
||||
}
|
||||
return mgotool.Count(ctx, g.coll, bson.M{"create_time": bson.M{"$lt": before}})
|
||||
}
|
||||
|
||||
func (g *GroupMgo) CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error) {
|
||||
//TODO implement me
|
||||
panic("implement me")
|
||||
}
|
||||
@ -0,0 +1,88 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
func NewGroupMember(db *mongo.Database) (relation.GroupMemberModelInterface, error) {
|
||||
return &GroupMemberMgo{coll: db.Collection("group_member")}, nil
|
||||
}
|
||||
|
||||
type GroupMemberMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) Create(ctx context.Context, groupMembers []*relation.GroupMemberModel) (err error) {
|
||||
return mgotool.InsertMany(ctx, g.coll, groupMembers)
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []string) (err error) {
|
||||
return mgotool.DeleteMany(ctx, g.coll, bson.M{"group_id": groupID, "user_id": bson.M{"$in": userIDs}})
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, userID string, roleLevel int32) error {
|
||||
return g.Update(ctx, groupID, userID, bson.M{"role_level": roleLevel})
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) Update(ctx context.Context, groupID string, userID string, data map[string]any) (err error) {
|
||||
return mgotool.UpdateOne(ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID}, bson.M{"$set": data}, true)
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) Find(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32) (groupMembers []*relation.GroupMemberModel, err error) {
|
||||
//TODO implement me
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) FindMemberUserID(ctx context.Context, groupID string) (userIDs []string, err error) {
|
||||
return mgotool.Find[string](ctx, g.coll, bson.M{"group_id": groupID}, options.Find().SetProjection(bson.M{"_id": 0, "user_id": 1}))
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) Take(ctx context.Context, groupID string, userID string) (groupMember *relation.GroupMemberModel, err error) {
|
||||
return mgotool.FindOne[*relation.GroupMemberModel](ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID})
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) TakeOwner(ctx context.Context, groupID string) (groupMember *relation.GroupMemberModel, err error) {
|
||||
return mgotool.FindOne[*relation.GroupMemberModel](ctx, g.coll, bson.M{"group_id": groupID, "role_level": constant.GroupOwner})
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) FindRoleLevelUserIDs(ctx context.Context, groupID string, roleLevel int32) ([]string, error) {
|
||||
return mgotool.Find[string](ctx, g.coll, bson.M{"group_id": groupID, "role_level": roleLevel}, options.Find().SetProjection(bson.M{"_id": 0, "user_id": 1}))
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) SearchMember(ctx context.Context, keyword string, groupID string, pagination pagination.Pagination) (total int64, groupList []*relation.GroupMemberModel, err error) {
|
||||
//TODO implement me
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) {
|
||||
return mgotool.Find[string](ctx, g.coll, bson.M{"user_id": userID}, options.Find().SetProjection(bson.M{"_id": 0, "group_id": 1}))
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) TakeGroupMemberNum(ctx context.Context, groupID string) (count int64, err error) {
|
||||
return mgotool.Count(ctx, g.coll, bson.M{"group_id": groupID})
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) {
|
||||
filter := bson.M{
|
||||
"user_id": userID,
|
||||
"role_level": bson.M{
|
||||
"$in": []int{constant.GroupOwner, constant.GroupAdmin},
|
||||
},
|
||||
}
|
||||
return mgotool.Find[string](ctx, g.coll, filter, options.Find().SetProjection(bson.M{"_id": 0, "group_id": 1}))
|
||||
}
|
||||
|
||||
func (g *GroupMemberMgo) IsUpdateRoleLevel(data map[string]any) bool {
|
||||
if len(data) == 0 {
|
||||
return false
|
||||
}
|
||||
_, ok := data["role_level"]
|
||||
return ok
|
||||
}
|
||||
@ -0,0 +1,46 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
)
|
||||
|
||||
func NewGroupRequestMgo(db *mongo.Database) (relation.GroupRequestModelInterface, error) {
|
||||
return &GroupRequestMgo{coll: db.Collection("group_request")}, nil
|
||||
}
|
||||
|
||||
type GroupRequestMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) Create(ctx context.Context, groupRequests []*relation.GroupRequestModel) (err error) {
|
||||
return mgotool.InsertMany(ctx, g.coll, groupRequests)
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) Delete(ctx context.Context, groupID string, userID string) (err error) {
|
||||
return mgotool.DeleteOne(ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID})
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) UpdateHandler(ctx context.Context, groupID string, userID string, handledMsg string, handleResult int32) (err error) {
|
||||
return mgotool.UpdateOne(ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID}, bson.M{"$set": bson.M{"handle_msg": handledMsg, "handle_result": handleResult}}, true)
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) Take(ctx context.Context, groupID string, userID string) (groupRequest *relation.GroupRequestModel, err error) {
|
||||
return mgotool.FindOne[*relation.GroupRequestModel](ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID})
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) FindGroupRequests(ctx context.Context, groupID string, userIDs []string) ([]*relation.GroupRequestModel, error) {
|
||||
return mgotool.Find[*relation.GroupRequestModel](ctx, g.coll, bson.M{"group_id": groupID, "user_id": bson.M{"$in": userIDs}})
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) Page(ctx context.Context, userID string, pagination pagination.Pagination) (total int64, groups []*relation.GroupRequestModel, err error) {
|
||||
return mgotool.FindPage[*relation.GroupRequestModel](ctx, g.coll, bson.M{"user_id": userID}, pagination)
|
||||
}
|
||||
|
||||
func (g *GroupRequestMgo) PageGroup(ctx context.Context, groupIDs []string, pagination pagination.Pagination) (total int64, groups []*relation.GroupRequestModel, err error) {
|
||||
return mgotool.FindPage[*relation.GroupRequestModel](ctx, g.coll, bson.M{"group_id": bson.M{"$in": groupIDs}}, pagination)
|
||||
}
|
||||
@ -0,0 +1,49 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewLogMongo(db *mongo.Database) (relation.LogInterface, error) {
|
||||
lm := &LogMgo{
|
||||
coll: db.Collection("log"),
|
||||
}
|
||||
return lm, nil
|
||||
}
|
||||
|
||||
type LogMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (l *LogMgo) Create(ctx context.Context, log []*relation.Log) error {
|
||||
return mgotool.InsertMany(ctx, l.coll, log)
|
||||
}
|
||||
|
||||
func (l *LogMgo) Search(ctx context.Context, keyword string, start time.Time, end time.Time, pagination pagination.Pagination) (int64, []*relation.Log, error) {
|
||||
filter := bson.M{"create_time": bson.M{"$gte": start, "$lte": end}}
|
||||
if keyword != "" {
|
||||
filter["user_id"] = bson.M{"$regex": keyword}
|
||||
}
|
||||
return mgotool.FindPage[*relation.Log](ctx, l.coll, filter, pagination, options.Find().SetSort(bson.M{"create_time": -1}))
|
||||
}
|
||||
|
||||
func (l *LogMgo) Delete(ctx context.Context, logID []string, userID string) error {
|
||||
if userID == "" {
|
||||
return mgotool.DeleteMany(ctx, l.coll, bson.M{"log_id": bson.M{"$in": logID}})
|
||||
}
|
||||
return mgotool.DeleteMany(ctx, l.coll, bson.M{"log_id": bson.M{"$in": logID}, "user_id": userID})
|
||||
}
|
||||
|
||||
func (l *LogMgo) Get(ctx context.Context, logIDs []string, userID string) ([]*relation.Log, error) {
|
||||
if userID == "" {
|
||||
return mgotool.Find[*relation.Log](ctx, l.coll, bson.M{"log_id": bson.M{"$in": logIDs}})
|
||||
}
|
||||
return mgotool.Find[*relation.Log](ctx, l.coll, bson.M{"log_id": bson.M{"$in": logIDs}, "user_id": userID})
|
||||
}
|
||||
@ -0,0 +1,198 @@
|
||||
package mgotool
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
|
||||
func basic[T any]() bool {
|
||||
var t T
|
||||
switch any(t).(type) {
|
||||
case int:
|
||||
case int8:
|
||||
case int16:
|
||||
case int32:
|
||||
case int64:
|
||||
case uint:
|
||||
case uint8:
|
||||
case uint16:
|
||||
case uint32:
|
||||
case uint64:
|
||||
case float32:
|
||||
case float64:
|
||||
case string:
|
||||
case []byte:
|
||||
default:
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func anes[T any](ts []T) []any {
|
||||
val := make([]any, len(ts))
|
||||
for i := range ts {
|
||||
val[i] = ts[i]
|
||||
}
|
||||
return val
|
||||
}
|
||||
|
||||
func findOptionToCountOption(opts []*options.FindOptions) *options.CountOptions {
|
||||
countOpt := options.Count()
|
||||
for _, opt := range opts {
|
||||
if opt.Skip != nil {
|
||||
countOpt.SetSkip(*opt.Skip)
|
||||
}
|
||||
if opt.Limit != nil {
|
||||
countOpt.SetLimit(*opt.Limit)
|
||||
}
|
||||
}
|
||||
return countOpt
|
||||
}
|
||||
|
||||
func InsertMany[T any](ctx context.Context, coll *mongo.Collection, val []T, opts ...*options.InsertManyOptions) error {
|
||||
_, err := coll.InsertMany(ctx, anes(val), opts...)
|
||||
if err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func UpdateOne(ctx context.Context, coll *mongo.Collection, filter any, update any, notMatchedErr bool, opts ...*options.UpdateOptions) error {
|
||||
res, err := coll.UpdateOne(ctx, filter, update, opts...)
|
||||
if err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
if notMatchedErr && res.MatchedCount == 0 {
|
||||
return errs.Wrap(mongo.ErrNoDocuments)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func UpdateMany(ctx context.Context, coll *mongo.Collection, filter any, update any, opts ...*options.UpdateOptions) (*mongo.UpdateResult, error) {
|
||||
res, err := coll.UpdateMany(ctx, filter, update, opts...)
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func Find[T any](ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.FindOptions) ([]T, error) {
|
||||
cur, err := coll.Find(ctx, filter, opts...)
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
defer cur.Close(ctx)
|
||||
var res []T
|
||||
if basic[T]() {
|
||||
var temp []map[string]T
|
||||
if err := cur.All(ctx, &temp); err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
res = make([]T, 0, len(temp))
|
||||
for _, m := range temp {
|
||||
if len(m) != 1 {
|
||||
return nil, errs.ErrInternalServer.Wrap("mongo find result len(m) != 1")
|
||||
}
|
||||
for _, t := range m {
|
||||
res = append(res, t)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if err := cur.All(ctx, &res); err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func FindOne[T any](ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.FindOneOptions) (res T, err error) {
|
||||
cur := coll.FindOne(ctx, filter, opts...)
|
||||
if err := cur.Err(); err != nil {
|
||||
return res, errs.Wrap(err)
|
||||
}
|
||||
if err := cur.Decode(&res); err != nil {
|
||||
return res, errs.Wrap(err)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func FindPage[T any](ctx context.Context, coll *mongo.Collection, filter any, pagination pagination.Pagination, opts ...*options.FindOptions) (int64, []T, error) {
|
||||
count, err := Count(ctx, coll, filter, findOptionToCountOption(opts))
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
if count == 0 || pagination == nil {
|
||||
return count, nil, nil
|
||||
}
|
||||
skip := int64(pagination.GetPageNumber()-1) * int64(pagination.GetShowNumber())
|
||||
if skip < 0 || skip >= count || pagination.GetShowNumber() <= 0 {
|
||||
return count, nil, nil
|
||||
}
|
||||
opt := options.Find().SetSkip(skip).SetLimit(int64(pagination.GetShowNumber()))
|
||||
res, err := Find[T](ctx, coll, filter, append(opts, opt)...)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
return count, res, nil
|
||||
}
|
||||
|
||||
func FindPageOnly[T any](ctx context.Context, coll *mongo.Collection, filter any, pagination pagination.Pagination, opts ...*options.FindOptions) ([]T, error) {
|
||||
skip := int64(pagination.GetPageNumber()-1) * int64(pagination.GetShowNumber())
|
||||
if skip < 0 || pagination.GetShowNumber() <= 0 {
|
||||
return nil, nil
|
||||
}
|
||||
opt := options.Find().SetSkip(skip).SetLimit(int64(pagination.GetShowNumber()))
|
||||
return Find[T](ctx, coll, filter, append(opts, opt)...)
|
||||
}
|
||||
|
||||
func Count(ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.CountOptions) (int64, error) {
|
||||
return coll.CountDocuments(ctx, filter, opts...)
|
||||
}
|
||||
|
||||
func Exist(ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.CountOptions) (bool, error) {
|
||||
opts = append(opts, options.Count().SetLimit(1))
|
||||
count, err := Count(ctx, coll, filter, opts...)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return count > 0, nil
|
||||
}
|
||||
|
||||
func DeleteOne(ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.DeleteOptions) error {
|
||||
if _, err := coll.DeleteOne(ctx, filter, opts...); err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func DeleteMany(ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.DeleteOptions) error {
|
||||
if _, err := coll.DeleteMany(ctx, filter, opts...); err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// TODO
|
||||
func Delete[T any](ctx context.Context, coll *mongo.Collection, filter any, opts ...*options.DeleteOptions) error {
|
||||
if _, err := coll.DeleteMany(ctx, filter, opts...); err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func Aggregate[T any](ctx context.Context, coll *mongo.Collection, pipeline any, opts ...*options.AggregateOptions) ([]T, error) {
|
||||
cur, err := coll.Aggregate(ctx, pipeline, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var ts []T
|
||||
if err := cur.All(ctx, &ts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ts, nil
|
||||
}
|
||||
@ -0,0 +1,45 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
func NewS3Mongo(db *mongo.Database) (relation.ObjectInfoModelInterface, error) {
|
||||
return &S3Mongo{
|
||||
coll: db.Collection("s3"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
type S3Mongo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (o *S3Mongo) SetObject(ctx context.Context, obj *relation.ObjectModel) error {
|
||||
filter := bson.M{"name": obj.Name, "engine": obj.Engine}
|
||||
update := bson.M{
|
||||
"name": obj.Name,
|
||||
"engine": obj.Engine,
|
||||
"key": obj.Key,
|
||||
"size": obj.Size,
|
||||
"content_type": obj.ContentType,
|
||||
"group": obj.Group,
|
||||
"create_time": obj.CreateTime,
|
||||
}
|
||||
return mgotool.UpdateOne(ctx, o.coll, filter, bson.M{"$set": update}, false, options.Update().SetUpsert(true))
|
||||
}
|
||||
|
||||
func (o *S3Mongo) Take(ctx context.Context, engine string, name string) (*relation.ObjectModel, error) {
|
||||
if engine == "" {
|
||||
return mgotool.FindOne[*relation.ObjectModel](ctx, o.coll, bson.M{"name": name})
|
||||
}
|
||||
return mgotool.FindOne[*relation.ObjectModel](ctx, o.coll, bson.M{"name": name, "engine": engine})
|
||||
}
|
||||
|
||||
func (o *S3Mongo) Delete(ctx context.Context, engine string, name string) error {
|
||||
return mgotool.DeleteOne(ctx, o.coll, bson.M{"name": name, "engine": engine})
|
||||
}
|
||||
@ -0,0 +1,71 @@
|
||||
package newmgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/newmgo/mgotool"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewUserMongo(db *mongo.Database) (relation.UserModelInterface, error) {
|
||||
return &UserMgo{
|
||||
coll: db.Collection("user"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
type UserMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (u *UserMgo) Create(ctx context.Context, users []*relation.UserModel) error {
|
||||
return mgotool.InsertMany(ctx, u.coll, users)
|
||||
}
|
||||
|
||||
func (u *UserMgo) UpdateByMap(ctx context.Context, userID string, args map[string]any) (err error) {
|
||||
if len(args) == 0 {
|
||||
return nil
|
||||
}
|
||||
return mgotool.UpdateOne(ctx, u.coll, bson.M{"user_id": userID}, bson.M{"$set": args}, true)
|
||||
}
|
||||
|
||||
func (u *UserMgo) Find(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error) {
|
||||
return mgotool.Find[*relation.UserModel](ctx, u.coll, bson.M{"user_id": bson.M{"$in": userIDs}})
|
||||
}
|
||||
|
||||
func (u *UserMgo) Take(ctx context.Context, userID string) (user *relation.UserModel, err error) {
|
||||
return mgotool.FindOne[*relation.UserModel](ctx, u.coll, bson.M{"user_id": userID})
|
||||
}
|
||||
|
||||
func (u *UserMgo) Page(ctx context.Context, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
|
||||
return mgotool.FindPage[*relation.UserModel](ctx, u.coll, bson.M{}, pagination)
|
||||
}
|
||||
|
||||
func (u *UserMgo) GetAllUserID(ctx context.Context, pagination pagination.Pagination) (int64, []string, error) {
|
||||
return mgotool.FindPage[string](ctx, u.coll, bson.M{}, pagination, options.Find().SetProjection(bson.M{"user_id": 1}))
|
||||
}
|
||||
|
||||
func (u *UserMgo) Exist(ctx context.Context, userID string) (exist bool, err error) {
|
||||
return mgotool.Exist(ctx, u.coll, bson.M{"user_id": userID})
|
||||
}
|
||||
|
||||
func (u *UserMgo) GetUserGlobalRecvMsgOpt(ctx context.Context, userID string) (opt int, err error) {
|
||||
return mgotool.FindOne[int](ctx, u.coll, bson.M{"user_id": userID}, options.FindOne().SetProjection(bson.M{"global_recv_msg_opt": 1}))
|
||||
}
|
||||
|
||||
func (u *UserMgo) CountTotal(ctx context.Context, before *time.Time) (count int64, err error) {
|
||||
return mgotool.Count(ctx, u.coll, bson.M{"create_time": bson.M{"$lt": before}})
|
||||
}
|
||||
|
||||
func (u *UserMgo) CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error) {
|
||||
//type Temp struct {
|
||||
// CreateTime time.Time `bson:"create_time"`
|
||||
// Number int64 `bson:"number"`
|
||||
//}
|
||||
//mgotool.Find(ctx, u.coll, bson.M{"create_time": bson.M{"$gte": start, "$lt": end}}, options.Find().SetProjection(bson.M{"create_time": 1}))
|
||||
panic("implement me")
|
||||
return nil, nil
|
||||
}
|
||||
@ -1,49 +1,49 @@
|
||||
package relation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
"github.com/OpenIMSDK/tools/ormutil"
|
||||
"gorm.io/gorm"
|
||||
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
)
|
||||
|
||||
type LogGorm struct {
|
||||
db *gorm.DB
|
||||
}
|
||||
|
||||
func (l *LogGorm) Create(ctx context.Context, log []*relationtb.Log) error {
|
||||
return errs.Wrap(l.db.WithContext(ctx).Create(log).Error)
|
||||
}
|
||||
|
||||
func (l *LogGorm) Search(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relationtb.Log, error) {
|
||||
db := l.db.WithContext(ctx).Where("create_time >= ?", start)
|
||||
if end.UnixMilli() != 0 {
|
||||
db = l.db.WithContext(ctx).Where("create_time <= ?", end)
|
||||
}
|
||||
db = db.Order("create_time desc")
|
||||
return ormutil.GormSearch[relationtb.Log](db, []string{"user_id"}, keyword, pageNumber, showNumber)
|
||||
}
|
||||
|
||||
func (l *LogGorm) Delete(ctx context.Context, logIDs []string, userID string) error {
|
||||
if userID == "" {
|
||||
return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Delete(&relationtb.Log{}).Error)
|
||||
}
|
||||
return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Delete(&relationtb.Log{}).Error)
|
||||
}
|
||||
|
||||
func (l *LogGorm) Get(ctx context.Context, logIDs []string, userID string) ([]*relationtb.Log, error) {
|
||||
var logs []*relationtb.Log
|
||||
if userID == "" {
|
||||
return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Find(&logs).Error)
|
||||
}
|
||||
return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Find(&logs).Error)
|
||||
}
|
||||
|
||||
func NewLogGorm(db *gorm.DB) relationtb.LogInterface {
|
||||
db.AutoMigrate(&relationtb.Log{})
|
||||
return &LogGorm{db: db}
|
||||
}
|
||||
//import (
|
||||
// "context"
|
||||
// "time"
|
||||
//
|
||||
// "github.com/OpenIMSDK/tools/errs"
|
||||
// "github.com/OpenIMSDK/tools/ormutil"
|
||||
// "gorm.io/gorm"
|
||||
//
|
||||
// relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
//)
|
||||
//
|
||||
//type LogGorm struct {
|
||||
// db *gorm.DB
|
||||
//}
|
||||
//
|
||||
//func (l *LogGorm) Create(ctx context.Context, log []*relationtb.Log) error {
|
||||
// return errs.Wrap(l.db.WithContext(ctx).Create(log).Error)
|
||||
//}
|
||||
//
|
||||
//func (l *LogGorm) Search(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relationtb.Log, error) {
|
||||
// db := l.db.WithContext(ctx).Where("create_time >= ?", start)
|
||||
// if end.UnixMilli() != 0 {
|
||||
// db = l.db.WithContext(ctx).Where("create_time <= ?", end)
|
||||
// }
|
||||
// db = db.Order("create_time desc")
|
||||
// return ormutil.GormSearch[relationtb.Log](db, []string{"user_id"}, keyword, pageNumber, showNumber)
|
||||
//}
|
||||
//
|
||||
//func (l *LogGorm) Delete(ctx context.Context, logIDs []string, userID string) error {
|
||||
// if userID == "" {
|
||||
// return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Delete(&relationtb.Log{}).Error)
|
||||
// }
|
||||
// return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Delete(&relationtb.Log{}).Error)
|
||||
//}
|
||||
//
|
||||
//func (l *LogGorm) Get(ctx context.Context, logIDs []string, userID string) ([]*relationtb.Log, error) {
|
||||
// var logs []*relationtb.Log
|
||||
// if userID == "" {
|
||||
// return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Find(&logs).Error)
|
||||
// }
|
||||
// return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Find(&logs).Error)
|
||||
//}
|
||||
//
|
||||
//func NewLogGorm(db *gorm.DB) relationtb.LogInterface {
|
||||
// db.AutoMigrate(&relationtb.Log{})
|
||||
// return &LogGorm{db: db}
|
||||
//}
|
||||
|
||||
@ -0,0 +1,19 @@
|
||||
package tx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/tx"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
)
|
||||
|
||||
func NewAuto(ctx context.Context, cli *mongo.Client) (tx.CtxTx, error) {
|
||||
var res map[string]any
|
||||
if err := cli.Database("admin").RunCommand(ctx, bson.M{"isMaster": 1}).Decode(&res); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, ok := res["setName"]; ok {
|
||||
return NewMongoTx(cli), nil
|
||||
}
|
||||
return NewInvalidTx(), nil
|
||||
}
|
||||
@ -0,0 +1,16 @@
|
||||
package tx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/tx"
|
||||
)
|
||||
|
||||
func NewInvalidTx() tx.CtxTx {
|
||||
return invalid{}
|
||||
}
|
||||
|
||||
type invalid struct{}
|
||||
|
||||
func (m invalid) Transaction(ctx context.Context, fn func(ctx context.Context) error) error {
|
||||
return fn(ctx)
|
||||
}
|
||||
@ -0,0 +1,28 @@
|
||||
package tx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/tx"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
)
|
||||
|
||||
func NewMongoTx(client *mongo.Client) tx.CtxTx {
|
||||
return &mongoTx{
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
|
||||
type mongoTx struct {
|
||||
client *mongo.Client
|
||||
}
|
||||
|
||||
func (m *mongoTx) Transaction(ctx context.Context, fn func(ctx context.Context) error) error {
|
||||
sess, err := m.client.StartSession()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = sess.WithTransaction(ctx, func(ctx mongo.SessionContext) (interface{}, error) {
|
||||
return nil, fn(ctx)
|
||||
})
|
||||
return err
|
||||
}
|
||||
@ -0,0 +1,6 @@
|
||||
package pagination
|
||||
|
||||
type Pagination interface {
|
||||
GetPageNumber() int32
|
||||
GetShowNumber() int32
|
||||
}
|
||||
@ -0,0 +1,87 @@
|
||||
package grouphash
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"github.com/OpenIMSDK/protocol/group"
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
)
|
||||
|
||||
func NewGroupHashFromGroupClient(x group.GroupClient) *GroupHash {
|
||||
return &GroupHash{
|
||||
getGroupAllUserIDs: func(ctx context.Context, groupID string) ([]string, error) {
|
||||
resp, err := x.GetGroupMemberUserIDs(ctx, &group.GetGroupMemberUserIDsReq{GroupID: groupID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.UserIDs, nil
|
||||
},
|
||||
getGroupMemberInfo: func(ctx context.Context, groupID string, userIDs []string) ([]*sdkws.GroupMemberFullInfo, error) {
|
||||
resp, err := x.GetGroupMembersInfo(ctx, &group.GetGroupMembersInfoReq{GroupID: groupID, UserIDs: userIDs})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.Members, nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func NewGroupHashFromGroupServer(x group.GroupServer) *GroupHash {
|
||||
return &GroupHash{
|
||||
getGroupAllUserIDs: func(ctx context.Context, groupID string) ([]string, error) {
|
||||
resp, err := x.GetGroupMemberUserIDs(ctx, &group.GetGroupMemberUserIDsReq{GroupID: groupID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.UserIDs, nil
|
||||
},
|
||||
getGroupMemberInfo: func(ctx context.Context, groupID string, userIDs []string) ([]*sdkws.GroupMemberFullInfo, error) {
|
||||
resp, err := x.GetGroupMembersInfo(ctx, &group.GetGroupMembersInfoReq{GroupID: groupID, UserIDs: userIDs})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.Members, nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
type GroupHash struct {
|
||||
getGroupAllUserIDs func(ctx context.Context, groupID string) ([]string, error)
|
||||
getGroupMemberInfo func(ctx context.Context, groupID string, userIDs []string) ([]*sdkws.GroupMemberFullInfo, error)
|
||||
}
|
||||
|
||||
func (gh *GroupHash) GetGroupHash(ctx context.Context, groupID string) (uint64, error) {
|
||||
userIDs, err := gh.getGroupAllUserIDs(ctx, groupID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var members []*sdkws.GroupMemberFullInfo
|
||||
if len(userIDs) > 0 {
|
||||
members, err = gh.getGroupMemberInfo(ctx, groupID, userIDs)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
utils.Sort(userIDs, true)
|
||||
}
|
||||
memberMap := utils.SliceToMap(members, func(e *sdkws.GroupMemberFullInfo) string {
|
||||
return e.UserID
|
||||
})
|
||||
res := make([]*sdkws.GroupMemberFullInfo, 0, len(members))
|
||||
for _, userID := range userIDs {
|
||||
member, ok := memberMap[userID]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
member.AppMangerLevel = 0
|
||||
res = append(res, member)
|
||||
}
|
||||
data, err := json.Marshal(res)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
sum := md5.Sum(data)
|
||||
return binary.BigEndian.Uint64(sum[:]), nil
|
||||
}
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue