feat: support incremental synchronization (#2379)
* fix: GroupApplicationAcceptedNotification * fix: GroupApplicationAcceptedNotification * fix: NotificationUserInfoUpdate * cicd: robot automated Change * fix: component * fix: getConversationInfo * feat: cron task * feat: cron task * feat: cron task * feat: cron task * feat: cron task * fix: minio config url recognition error * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * friend incr sync * friend incr sync * friend incr sync * friend incr sync * friend incr sync * mage * optimization version log * optimization version log * sync * sync * sync * group sync * sync option * sync option * refactor: replace `friend` package with `realtion`. * refactor: update lastest commit to relation. * sync option * sync option * sync option * sync * sync * go.mod * update: go mod * refactor: change incremental to full * feat: get full friend user ids * feat: api and config * group version * merge * fix: sort by id avoid unstable sort friends. * group * group * group * fix: sort by id avoid unstable sort friends. * fix: sort by id avoid unstable sort friends. * fix: sort by id avoid unstable sort friends. * user version * fix: sort by id avoid unstable sort friends. * test: test log add. * test: debug log remove. * fix: transfer group owner incr version more than 1. * fix: add condition to kick owner. * feat: replace resp nil * feat: replace nil * fix: delete cache of max group joined version avoid sync joined group failed. * fix: nil * fix: delete cache of max group joined version avoid sync joined group failed. * fix: delete cache of max group joined version avoid sync joined group failed. * return group information for any changes * online cache --------- Co-authored-by: withchao <withchao@users.noreply.github.com> Co-authored-by: Monet Lee <monet_lee@163.com> Co-authored-by: OpenIM-Gordon <46924906+FGadvancer@users.noreply.github.com> Co-authored-by: icey-yu <1186114839@qq.com>pull/2389/head
chao
9 months ago
committed by
GitHub
parent
fe7c029c2a
commit
88c0d5f5ad
@ -0,0 +1,78 @@
|
||||
package friend
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/protocol/relation"
|
||||
)
|
||||
|
||||
func (s *friendServer) NotificationUserInfoUpdate(ctx context.Context, req *relation.NotificationUserInfoUpdateReq) (*relation.NotificationUserInfoUpdateResp, error) {
|
||||
userIDs, err := s.db.FindFriendUserIDs(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, userID := range userIDs {
|
||||
if err := s.db.OwnerIncrVersion(ctx, userID, []string{req.UserID}, model.VersionStateUpdate); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
for _, userID := range userIDs {
|
||||
s.notificationSender.FriendInfoUpdatedNotification(ctx, req.UserID, userID)
|
||||
}
|
||||
return &relation.NotificationUserInfoUpdateResp{}, nil
|
||||
}
|
||||
|
||||
func (s *friendServer) GetFullFriendUserIDs(ctx context.Context, req *relation.GetFullFriendUserIDsReq) (*relation.GetFullFriendUserIDsResp, error) {
|
||||
vl, err := s.db.FindMaxFriendVersionCache(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
userIDs, err := s.db.FindFriendUserIDs(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
idHash := hashutil.IdHash(userIDs)
|
||||
if req.IdHash == idHash {
|
||||
userIDs = nil
|
||||
}
|
||||
return &relation.GetFullFriendUserIDsResp{
|
||||
Version: idHash,
|
||||
VersionID: vl.ID.Hex(),
|
||||
Equal: req.IdHash == idHash,
|
||||
UserIDs: userIDs,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.GetIncrementalFriendsReq) (*relation.GetIncrementalFriendsResp, error) {
|
||||
if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
opt := incrversion.Option[*sdkws.FriendInfo, relation.GetIncrementalFriendsResp]{
|
||||
Ctx: ctx,
|
||||
VersionKey: req.UserID,
|
||||
VersionID: req.VersionID,
|
||||
VersionNumber: req.Version,
|
||||
Version: s.db.FindFriendIncrVersion,
|
||||
CacheMaxVersion: s.db.FindMaxFriendVersionCache,
|
||||
Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) {
|
||||
return s.getFriend(ctx, req.UserID, ids)
|
||||
},
|
||||
ID: func(elem *sdkws.FriendInfo) string { return elem.FriendUser.UserID },
|
||||
Resp: func(version *model.VersionLog, deleteIds []string, insertList, updateList []*sdkws.FriendInfo, full bool) *relation.GetIncrementalFriendsResp {
|
||||
return &relation.GetIncrementalFriendsResp{
|
||||
VersionID: version.ID.Hex(),
|
||||
Version: uint64(version.Version),
|
||||
Full: full,
|
||||
Delete: deleteIds,
|
||||
Insert: insertList,
|
||||
Update: updateList,
|
||||
}
|
||||
},
|
||||
}
|
||||
return opt.Build()
|
||||
}
|
||||
@ -0,0 +1,149 @@
|
||||
package group
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
pbgroup "github.com/openimsdk/protocol/group"
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
"slices"
|
||||
)
|
||||
|
||||
func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) {
|
||||
vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
userIDs, err := s.db.FindGroupMemberUserID(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
idHash := hashutil.IdHash(userIDs)
|
||||
if req.IdHash == idHash {
|
||||
userIDs = nil
|
||||
}
|
||||
return &pbgroup.GetFullGroupMemberUserIDsResp{
|
||||
Version: idHash,
|
||||
VersionID: vl.ID.Hex(),
|
||||
Equal: req.IdHash == idHash,
|
||||
UserIDs: userIDs,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *groupServer) GetFullJoinGroupIDs(ctx context.Context, req *pbgroup.GetFullJoinGroupIDsReq) (*pbgroup.GetFullJoinGroupIDsResp, error) {
|
||||
vl, err := s.db.FindMaxJoinGroupVersionCache(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
groupIDs, err := s.db.FindJoinGroupID(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
idHash := hashutil.IdHash(groupIDs)
|
||||
if req.IdHash == idHash {
|
||||
groupIDs = nil
|
||||
}
|
||||
return &pbgroup.GetFullJoinGroupIDsResp{
|
||||
Version: idHash,
|
||||
VersionID: vl.ID.Hex(),
|
||||
Equal: req.IdHash == idHash,
|
||||
GroupIDs: groupIDs,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgroup.GetIncrementalGroupMemberReq) (*pbgroup.GetIncrementalGroupMemberResp, error) {
|
||||
group, err := s.db.TakeGroup(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if group.Status == constant.GroupStatusDismissed {
|
||||
return nil, servererrs.ErrDismissedAlready.Wrap()
|
||||
}
|
||||
var hasGroupUpdate bool
|
||||
opt := incrversion.Option[*sdkws.GroupMemberFullInfo, pbgroup.GetIncrementalGroupMemberResp]{
|
||||
Ctx: ctx,
|
||||
VersionKey: req.GroupID,
|
||||
VersionID: req.VersionID,
|
||||
VersionNumber: req.Version,
|
||||
Version: 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
|
||||
})
|
||||
if vl.LogLen > 0 {
|
||||
hasGroupUpdate = true
|
||||
}
|
||||
return vl, nil
|
||||
},
|
||||
CacheMaxVersion: s.db.FindMaxGroupMemberVersionCache,
|
||||
Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) {
|
||||
return s.getGroupMembersInfo(ctx, req.GroupID, ids)
|
||||
},
|
||||
ID: func(elem *sdkws.GroupMemberFullInfo) string { return elem.UserID },
|
||||
Resp: func(version *model.VersionLog, delIDs []string, insertList, updateList []*sdkws.GroupMemberFullInfo, full bool) *pbgroup.GetIncrementalGroupMemberResp {
|
||||
return &pbgroup.GetIncrementalGroupMemberResp{
|
||||
VersionID: version.ID.Hex(),
|
||||
Version: uint64(version.Version),
|
||||
Full: full,
|
||||
Delete: delIDs,
|
||||
Insert: insertList,
|
||||
Update: updateList,
|
||||
}
|
||||
},
|
||||
}
|
||||
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) {
|
||||
if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
opt := incrversion.Option[*sdkws.GroupInfo, pbgroup.GetIncrementalJoinGroupResp]{
|
||||
Ctx: ctx,
|
||||
VersionKey: req.UserID,
|
||||
VersionID: req.VersionID,
|
||||
VersionNumber: req.Version,
|
||||
Version: s.db.FindJoinIncrVersion,
|
||||
CacheMaxVersion: s.db.FindMaxJoinGroupVersionCache,
|
||||
Find: s.getGroupsInfo,
|
||||
ID: func(elem *sdkws.GroupInfo) string { return elem.GroupID },
|
||||
Resp: func(version *model.VersionLog, delIDs []string, insertList, updateList []*sdkws.GroupInfo, full bool) *pbgroup.GetIncrementalJoinGroupResp {
|
||||
return &pbgroup.GetIncrementalJoinGroupResp{
|
||||
VersionID: version.ID.Hex(),
|
||||
Version: uint64(version.Version),
|
||||
Full: full,
|
||||
Delete: delIDs,
|
||||
Insert: insertList,
|
||||
Update: updateList,
|
||||
}
|
||||
},
|
||||
}
|
||||
return opt.Build()
|
||||
}
|
||||
@ -0,0 +1,156 @@
|
||||
package incrversion
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
)
|
||||
|
||||
//func Limit(maxSync int, version uint64) int {
|
||||
// if version == 0 {
|
||||
// return 0
|
||||
// }
|
||||
// return maxSync
|
||||
//}
|
||||
|
||||
const syncLimit = 200
|
||||
|
||||
const (
|
||||
tagQuery = iota + 1
|
||||
tagFull
|
||||
tageEqual
|
||||
)
|
||||
|
||||
type Option[A, B any] struct {
|
||||
Ctx context.Context
|
||||
VersionKey string
|
||||
VersionID string
|
||||
VersionNumber uint64
|
||||
//SyncLimit int
|
||||
CacheMaxVersion func(ctx context.Context, dId string) (*model.VersionLog, error)
|
||||
Version func(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error)
|
||||
//SortID func(ctx context.Context, dId string) ([]string, error)
|
||||
Find func(ctx context.Context, ids []string) ([]A, error)
|
||||
ID func(elem A) string
|
||||
Resp func(version *model.VersionLog, deleteIds []string, insertList, updateList []A, full bool) *B
|
||||
}
|
||||
|
||||
func (o *Option[A, B]) newError(msg string) error {
|
||||
return errs.ErrInternalServer.WrapMsg(msg)
|
||||
}
|
||||
|
||||
func (o *Option[A, B]) check() error {
|
||||
if o.Ctx == nil {
|
||||
return o.newError("opt ctx is nil")
|
||||
}
|
||||
if o.VersionKey == "" {
|
||||
return o.newError("versionKey is empty")
|
||||
}
|
||||
//if o.SyncLimit <= 0 {
|
||||
// return o.newError("invalid synchronization quantity")
|
||||
//}
|
||||
if o.Version == nil {
|
||||
return o.newError("func version is nil")
|
||||
}
|
||||
//if o.SortID == nil {
|
||||
// return o.newError("func allID is nil")
|
||||
//}
|
||||
if o.Find == nil {
|
||||
return o.newError("func find is nil")
|
||||
}
|
||||
if o.ID == nil {
|
||||
return o.newError("func id is nil")
|
||||
}
|
||||
if o.Resp == nil {
|
||||
return o.newError("func resp is nil")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (o *Option[A, B]) validVersion() bool {
|
||||
objID, err := primitive.ObjectIDFromHex(o.VersionID)
|
||||
return err == nil && (!objID.IsZero()) && o.VersionNumber > 0
|
||||
}
|
||||
|
||||
func (o *Option[A, B]) equalID(objID primitive.ObjectID) bool {
|
||||
return o.VersionID == objID.Hex()
|
||||
}
|
||||
|
||||
func (o *Option[A, B]) getVersion(tag *int) (*model.VersionLog, error) {
|
||||
if o.CacheMaxVersion == nil {
|
||||
if o.validVersion() {
|
||||
*tag = tagQuery
|
||||
return o.Version(o.Ctx, o.VersionKey, uint(o.VersionNumber), syncLimit)
|
||||
}
|
||||
*tag = tagFull
|
||||
return o.Version(o.Ctx, o.VersionKey, 0, 0)
|
||||
} else {
|
||||
cache, err := o.CacheMaxVersion(o.Ctx, o.VersionKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !o.validVersion() {
|
||||
*tag = tagFull
|
||||
return cache, nil
|
||||
}
|
||||
if !o.equalID(cache.ID) {
|
||||
*tag = tagFull
|
||||
return cache, nil
|
||||
}
|
||||
if o.VersionNumber == uint64(cache.Version) {
|
||||
*tag = tageEqual
|
||||
return cache, nil
|
||||
}
|
||||
*tag = tagQuery
|
||||
return o.Version(o.Ctx, o.VersionKey, uint(o.VersionNumber), syncLimit)
|
||||
}
|
||||
}
|
||||
|
||||
func (o *Option[A, B]) Build() (*B, error) {
|
||||
if err := o.check(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var tag int
|
||||
version, err := o.getVersion(&tag)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var full bool
|
||||
switch tag {
|
||||
case tagQuery:
|
||||
full = version.ID.Hex() != o.VersionID || uint64(version.Version) < o.VersionNumber || len(version.Logs) != version.LogLen
|
||||
case tagFull:
|
||||
full = true
|
||||
case tageEqual:
|
||||
full = false
|
||||
default:
|
||||
panic(fmt.Errorf("undefined tag %d", tag))
|
||||
}
|
||||
var (
|
||||
insertIds []string
|
||||
deleteIds []string
|
||||
updateIds []string
|
||||
)
|
||||
if !full {
|
||||
insertIds, deleteIds, updateIds = version.DeleteAndChangeIDs()
|
||||
}
|
||||
var (
|
||||
insertList []A
|
||||
updateList []A
|
||||
)
|
||||
if len(insertIds) > 0 {
|
||||
insertList, err = o.Find(o.Ctx, insertIds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if len(updateIds) > 0 {
|
||||
updateList, err = o.Find(o.Ctx, updateIds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return o.Resp(version, deleteIds, insertList, updateList, full), nil
|
||||
}
|
||||
@ -0,0 +1,265 @@
|
||||
package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/versionctx"
|
||||
"github.com/openimsdk/tools/db/mongoutil"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewVersionLog(coll *mongo.Collection) (database.VersionLog, error) {
|
||||
lm := &VersionLogMgo{coll: coll}
|
||||
if lm.initIndex(context.Background()) != nil {
|
||||
return nil, errs.ErrInternalServer.WrapMsg("init index failed", "coll", coll.Name())
|
||||
}
|
||||
return lm, nil
|
||||
}
|
||||
|
||||
type VersionLogMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) initIndex(ctx context.Context) error {
|
||||
_, err := l.coll.Indexes().CreateOne(ctx, mongo.IndexModel{
|
||||
Keys: bson.M{
|
||||
"d_id": 1,
|
||||
},
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) IncrVersion(ctx context.Context, dId string, eIds []string, state int32) error {
|
||||
_, err := l.IncrVersionResult(ctx, dId, eIds, state)
|
||||
return err
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) IncrVersionResult(ctx context.Context, dId string, eIds []string, state int32) (*model.VersionLog, error) {
|
||||
vl, err := l.incrVersionResult(ctx, dId, eIds, state)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
versionctx.GetVersionLog(ctx).Append(versionctx.Collection{
|
||||
Name: l.coll.Name(),
|
||||
Doc: vl,
|
||||
})
|
||||
return vl, nil
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) incrVersionResult(ctx context.Context, dId string, eIds []string, state int32) (*model.VersionLog, error) {
|
||||
if len(eIds) == 0 {
|
||||
return nil, errs.ErrArgs.WrapMsg("elem id is empty", "dId", dId)
|
||||
}
|
||||
now := time.Now()
|
||||
if res, err := l.writeLogBatch2(ctx, dId, eIds, state, now); err == nil {
|
||||
return res, nil
|
||||
} else if !errors.Is(err, mongo.ErrNoDocuments) {
|
||||
return nil, err
|
||||
}
|
||||
if res, err := l.initDoc(ctx, dId, eIds, state, now); err == nil {
|
||||
return res, nil
|
||||
} else if !mongo.IsDuplicateKeyError(err) {
|
||||
return nil, err
|
||||
}
|
||||
return l.writeLogBatch2(ctx, dId, eIds, state, now)
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) initDoc(ctx context.Context, dId string, eIds []string, state int32, now time.Time) (*model.VersionLog, error) {
|
||||
wl := model.VersionLogTable{
|
||||
ID: primitive.NewObjectID(),
|
||||
DID: dId,
|
||||
Logs: make([]model.VersionLogElem, 0, len(eIds)),
|
||||
Version: database.FirstVersion,
|
||||
Deleted: database.DefaultDeleteVersion,
|
||||
LastUpdate: now,
|
||||
}
|
||||
for _, eId := range eIds {
|
||||
wl.Logs = append(wl.Logs, model.VersionLogElem{
|
||||
EID: eId,
|
||||
State: state,
|
||||
Version: database.FirstVersion,
|
||||
LastUpdate: now,
|
||||
})
|
||||
}
|
||||
if _, err := l.coll.InsertOne(ctx, &wl); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return wl.VersionLog(), nil
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) writeLogBatch2(ctx context.Context, dId string, eIds []string, state int32, now time.Time) (*model.VersionLog, error) {
|
||||
if eIds == nil {
|
||||
eIds = []string{}
|
||||
}
|
||||
filter := bson.M{
|
||||
"d_id": dId,
|
||||
}
|
||||
elems := make([]bson.M, 0, len(eIds))
|
||||
for _, eId := range eIds {
|
||||
elems = append(elems, bson.M{
|
||||
"e_id": eId,
|
||||
"version": "$version",
|
||||
"state": state,
|
||||
"last_update": now,
|
||||
})
|
||||
}
|
||||
pipeline := []bson.M{
|
||||
{
|
||||
"$addFields": bson.M{
|
||||
"delete_e_ids": eIds,
|
||||
},
|
||||
},
|
||||
{
|
||||
"$set": bson.M{
|
||||
"version": bson.M{"$add": []any{"$version", 1}},
|
||||
"last_update": now,
|
||||
},
|
||||
},
|
||||
{
|
||||
"$set": bson.M{
|
||||
"logs": bson.M{
|
||||
"$filter": bson.M{
|
||||
"input": "$logs",
|
||||
"as": "log",
|
||||
"cond": bson.M{
|
||||
"$not": bson.M{
|
||||
"$in": []any{"$$log.e_id", "$delete_e_ids"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"$set": bson.M{
|
||||
"logs": bson.M{
|
||||
"$concatArrays": []any{
|
||||
"$logs",
|
||||
elems,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"$unset": "delete_e_ids",
|
||||
},
|
||||
}
|
||||
opt := options.FindOneAndUpdate().SetUpsert(false).SetReturnDocument(options.After).SetProjection(bson.M{"logs": 0})
|
||||
return mongoutil.FindOneAndUpdate[*model.VersionLog](ctx, l.coll, filter, pipeline, opt)
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) findDoc(ctx context.Context, dId string) (*model.VersionLog, error) {
|
||||
vl, err := mongoutil.FindOne[*model.VersionLogTable](ctx, l.coll, bson.M{"d_id": dId}, options.FindOne().SetProjection(bson.M{"logs": 0}))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return vl.VersionLog(), nil
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error) {
|
||||
if wl, err := l.findChangeLog(ctx, dId, version, limit); err == nil {
|
||||
return wl, nil
|
||||
} else if !errors.Is(err, mongo.ErrNoDocuments) {
|
||||
return nil, err
|
||||
}
|
||||
log.ZDebug(ctx, "init doc", "dId", dId)
|
||||
if res, err := l.initDoc(ctx, dId, nil, 0, time.Now()); err == nil {
|
||||
log.ZDebug(ctx, "init doc success", "dId", dId)
|
||||
return res, nil
|
||||
} else if mongo.IsDuplicateKeyError(err) {
|
||||
return l.findChangeLog(ctx, dId, version, limit)
|
||||
} else {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) findChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error) {
|
||||
if version == 0 && limit == 0 {
|
||||
return l.findDoc(ctx, dId)
|
||||
}
|
||||
pipeline := []bson.M{
|
||||
{
|
||||
"$match": bson.M{
|
||||
"d_id": dId,
|
||||
},
|
||||
},
|
||||
{
|
||||
"$addFields": bson.M{
|
||||
"logs": bson.M{
|
||||
"$cond": bson.M{
|
||||
"if": bson.M{
|
||||
"$or": []bson.M{
|
||||
{"$lt": []any{"$version", version}},
|
||||
{"$gte": []any{"$deleted", version}},
|
||||
},
|
||||
},
|
||||
"then": []any{},
|
||||
"else": "$logs",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"$addFields": bson.M{
|
||||
"logs": bson.M{
|
||||
"$filter": bson.M{
|
||||
"input": "$logs",
|
||||
"as": "l",
|
||||
"cond": bson.M{
|
||||
"$gt": []any{"$$l.version", version},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"$addFields": bson.M{
|
||||
"log_len": bson.M{"$size": "$logs"},
|
||||
},
|
||||
},
|
||||
{
|
||||
"$addFields": bson.M{
|
||||
"logs": bson.M{
|
||||
"$cond": bson.M{
|
||||
"if": bson.M{
|
||||
"$gt": []any{"$log_len", limit},
|
||||
},
|
||||
"then": []any{},
|
||||
"else": "$logs",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
if limit <= 0 {
|
||||
pipeline = pipeline[:len(pipeline)-1]
|
||||
}
|
||||
vl, err := mongoutil.Aggregate[*model.VersionLog](ctx, l.coll, pipeline)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(vl) == 0 {
|
||||
return nil, mongo.ErrNoDocuments
|
||||
}
|
||||
return vl[0], nil
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error {
|
||||
return mongoutil.DeleteMany(ctx, l.coll, bson.M{
|
||||
"last_update": bson.M{
|
||||
"$lt": deadline,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func (l *VersionLogMgo) Delete(ctx context.Context, dId string) error {
|
||||
return mongoutil.DeleteOne(ctx, l.coll, bson.M{"d_id": dId})
|
||||
}
|
||||
@ -0,0 +1,39 @@
|
||||
package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func Result[V any](val V, err error) V {
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return val
|
||||
}
|
||||
|
||||
func Check(err error) {
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestName(t *testing.T) {
|
||||
cli := Result(mongo.Connect(context.Background(), options.Client().ApplyURI("mongodb://openIM:openIM123@172.16.8.48:37017/openim_v3?maxPoolSize=100").SetConnectTimeout(5*time.Second)))
|
||||
coll := cli.Database("openim_v3").Collection("version_test")
|
||||
tmp, err := NewVersionLog(coll)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
vl := tmp.(*VersionLogMgo)
|
||||
res, err := vl.writeLogBatch2(context.Background(), "100", []string{"1000", "1001", "1003"}, model.VersionStateInsert, time.Now())
|
||||
if err != nil {
|
||||
t.Log(err)
|
||||
return
|
||||
}
|
||||
t.Logf("%+v", res)
|
||||
}
|
||||
@ -0,0 +1,17 @@
|
||||
package database
|
||||
|
||||
const (
|
||||
BlackName = "black"
|
||||
ConversationName = "conversation"
|
||||
FriendName = "friend"
|
||||
FriendVersionName = "friend_version"
|
||||
FriendRequestName = "friend_request"
|
||||
GroupName = "group"
|
||||
GroupMemberName = "group_member"
|
||||
GroupMemberVersionName = "group_member_version"
|
||||
GroupJoinVersionName = "group_join_version"
|
||||
GroupRequestName = "group_request"
|
||||
LogName = "log"
|
||||
ObjectName = "s3"
|
||||
UserName = "user"
|
||||
)
|
||||
@ -0,0 +1,19 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
FirstVersion = 1
|
||||
DefaultDeleteVersion = 0
|
||||
)
|
||||
|
||||
type VersionLog interface {
|
||||
IncrVersion(ctx context.Context, dId string, eIds []string, state int32) error
|
||||
FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error)
|
||||
DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error
|
||||
Delete(ctx context.Context, dId string) error
|
||||
}
|
||||
@ -0,0 +1,69 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
VersionStateInsert = iota + 1
|
||||
VersionStateDelete
|
||||
VersionStateUpdate
|
||||
)
|
||||
|
||||
type VersionLogElem struct {
|
||||
EID string `bson:"e_id"`
|
||||
State int32 `bson:"state"`
|
||||
Version uint `bson:"version"`
|
||||
LastUpdate time.Time `bson:"last_update"`
|
||||
}
|
||||
|
||||
type VersionLogTable struct {
|
||||
ID primitive.ObjectID `bson:"_id"`
|
||||
DID string `bson:"d_id"`
|
||||
Logs []VersionLogElem `bson:"logs"`
|
||||
Version uint `bson:"version"`
|
||||
Deleted uint `bson:"deleted"`
|
||||
LastUpdate time.Time `bson:"last_update"`
|
||||
}
|
||||
|
||||
func (v *VersionLogTable) VersionLog() *VersionLog {
|
||||
return &VersionLog{
|
||||
ID: v.ID,
|
||||
DID: v.DID,
|
||||
Logs: v.Logs,
|
||||
Version: v.Version,
|
||||
Deleted: v.Deleted,
|
||||
LastUpdate: v.LastUpdate,
|
||||
LogLen: len(v.Logs),
|
||||
}
|
||||
}
|
||||
|
||||
type VersionLog struct {
|
||||
ID primitive.ObjectID `bson:"_id"`
|
||||
DID string `bson:"d_id"`
|
||||
Logs []VersionLogElem `bson:"logs"`
|
||||
Version uint `bson:"version"`
|
||||
Deleted uint `bson:"deleted"`
|
||||
LastUpdate time.Time `bson:"last_update"`
|
||||
LogLen int `bson:"log_len"`
|
||||
}
|
||||
|
||||
func (v *VersionLog) DeleteAndChangeIDs() (insertIds, deleteIds, updateIds []string) {
|
||||
for _, l := range v.Logs {
|
||||
switch l.State {
|
||||
case VersionStateInsert:
|
||||
insertIds = append(insertIds, l.EID)
|
||||
case VersionStateDelete:
|
||||
deleteIds = append(deleteIds, l.EID)
|
||||
case VersionStateUpdate:
|
||||
updateIds = append(updateIds, l.EID)
|
||||
default:
|
||||
log.ZError(context.Background(), "invalid version status found", errors.New("dirty database data"), "objID", v.ID.Hex(), "did", v.DID, "elem", l)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
@ -0,0 +1,14 @@
|
||||
package versionctx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
func EnableVersionCtx() grpc.ServerOption {
|
||||
return grpc.ChainUnaryInterceptor(enableVersionCtxInterceptor)
|
||||
}
|
||||
|
||||
func enableVersionCtxInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
return handler(WithVersionLog(ctx), req)
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
package versionctx
|
||||
|
||||
import (
|
||||
"context"
|
||||
tablerelation "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type Collection struct {
|
||||
Name string
|
||||
Doc *tablerelation.VersionLog
|
||||
}
|
||||
|
||||
type versionKey struct{}
|
||||
|
||||
func WithVersionLog(ctx context.Context) context.Context {
|
||||
return context.WithValue(ctx, versionKey{}, &VersionLog{})
|
||||
}
|
||||
|
||||
func GetVersionLog(ctx context.Context) *VersionLog {
|
||||
if v, ok := ctx.Value(versionKey{}).(*VersionLog); ok {
|
||||
return v
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type VersionLog struct {
|
||||
lock sync.Mutex
|
||||
data []Collection
|
||||
}
|
||||
|
||||
func (v *VersionLog) Append(data ...Collection) {
|
||||
if v == nil || len(data) == 0 {
|
||||
return
|
||||
}
|
||||
v.lock.Lock()
|
||||
defer v.lock.Unlock()
|
||||
v.data = append(v.data, data...)
|
||||
}
|
||||
|
||||
func (v *VersionLog) Get() []Collection {
|
||||
if v == nil {
|
||||
return nil
|
||||
}
|
||||
v.lock.Lock()
|
||||
defer v.lock.Unlock()
|
||||
return v.data
|
||||
}
|
||||
@ -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[:])
|
||||
}
|
||||
Loading…
Reference in new issue