IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-17 07:19:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

proto modify

Browse Source
This commit is contained in:
wangchuxiao
2023-03-01 15:32:26 +08:00
parent 9ecd43af73
commit 8d0148c4ba
33 changed files with 721 additions and 1338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/db/controller/auth.go
+39 -10
View File
@@ -1,32 +1,61 @@
package controller
import (
"OpenIM/pkg/common/constant"
"OpenIM/pkg/common/db/cache"
"OpenIM/pkg/common/tokenverify"
"OpenIM/pkg/utils"
"context"
"github.com/go-redis/redis/v8"
"github.com/golang-jwt/jwt/v4"
)
type AuthInterface interface {
type AuthDatabase interface {
//结果为空 不返回错误
GetTokensWithoutError(ctx context.Context, userID, platform string) (map[string]int, error)
//创建token
CreateToken(ctx context.Context, userID string, platform string) (string, error)
}
type AuthController struct {
database *cache.TokenRedis
type authDatabase struct {
cache cache.Cache
accessSecret string
accessExpire int64
}
func NewAuthController(rdb redis.UniversalClient, accessSecret string, accessExpire int64) *AuthController {
return &AuthController{database: cache.NewTokenRedis(cache.NewRedisClient(rdb), accessSecret, accessExpire)}
func NewAuthDatabase(cache cache.Cache, accessSecret string, accessExpire int64) AuthDatabase {
return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire}
}
// 结果为空 不返回错误
func (a *AuthController) GetTokensWithoutError(ctx context.Context, userID, platform string) (map[string]int, error) {
return a.database.GetTokensWithoutError(ctx, userID, platform)
func (a *authDatabase) GetTokensWithoutError(ctx context.Context, userID, platform string) (map[string]int, error) {
return a.cache.GetTokensWithoutError(ctx, userID, platform)
}
// 创建token
func (a *AuthController) CreateToken(ctx context.Context, userID string, platform string) (string, error) {
return a.database.CreateToken(ctx, userID, platform)
func (a *authDatabase) CreateToken(ctx context.Context, userID string, platform string) (string, error) {
tokens, err := a.cache.GetTokensWithoutError(ctx, userID, platform)
if err != nil {
return "", err
}
var deleteTokenKey []string
for k, v := range tokens {
_, err = tokenverify.GetClaimFromToken(k)
if err != nil || v != constant.NormalToken {
deleteTokenKey = append(deleteTokenKey, k)
}
}
if len(deleteTokenKey) != 0 {
err := a.cache.DeleteTokenByUidPid(ctx, userID, platform, deleteTokenKey)
if err != nil {
return "", err
}
}
claims := tokenverify.BuildClaims(userID, platform, a.accessExpire)
token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
tokenString, err := token.SignedString([]byte(a.accessSecret))
if err != nil {
return "", utils.Wrap(err, "")
}
return tokenString, a.cache.AddTokenFlag(ctx, userID, constant.PlatformNameToID(platform), tokenString, constant.NormalToken)
}
pkg/common/db/controller/conversation.go
+2 -2
View File
@@ -11,7 +11,7 @@ import (
"encoding/json"
)
type ConversationDataBaseInterface interface {
type ConversationDatabase interface {
//GetUserIDExistConversation 获取拥有该会话的的用户ID列表
GetUserIDExistConversation(ctx context.Context, userIDList []string, conversationID string) ([]string, error)
//UpdateUserConversationFiled 更新用户该会话的属性信息
@@ -32,7 +32,7 @@ type ConversationDataBaseInterface interface {
SetUsersConversationFiledTx(ctx context.Context, userIDList []string, conversation *relationTb.ConversationModel, filedMap map[string]interface{}) error
}
func NewConversationDatabase(conversation relation.Conversation, cache cache.ConversationCache, tx tx.Tx) ConversationDataBaseInterface {
func NewConversationDatabase(conversation relation.Conversation, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
return &ConversationDataBase{
conversationDB: conversation,
cache: cache,
pkg/common/db/controller/extend_msg.go
+19 -66
View File
@@ -4,11 +4,9 @@ import (
unRelationTb "OpenIM/pkg/common/db/table/unrelation"
"OpenIM/pkg/proto/sdkws"
"context"
"github.com/go-redis/redis/v8"
"go.mongodb.org/mongo-driver/mongo"
)
type ExtendMsgInterface interface {
type ExtendMsgDatabase interface {
CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error
GetAllExtendMsgSet(ctx context.Context, ID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error)
GetExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error)
@@ -18,82 +16,37 @@ type ExtendMsgInterface interface {
GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error)
}
type ExtendMsgController struct {
database ExtendMsgDatabase
type extendMsgDatabase struct {
db unRelationTb.ExtendMsgSetModelInterface
}
func NewExtendMsgController(mgo *mongo.Client, rdb redis.UniversalClient) *ExtendMsgController {
return &ExtendMsgController{}
func NewExtendMsgDatabase() ExtendMsgDatabase {
return &extendMsgDatabase{}
}
func (e *ExtendMsgController) CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error {
return e.database.CreateExtendMsgSet(ctx, set)
func (e *extendMsgDatabase) CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error {
return e.db.CreateExtendMsgSet(ctx, set)
}
func (e *ExtendMsgController) GetAllExtendMsgSet(ctx context.Context, ID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error) {
return e.GetAllExtendMsgSet(ctx, ID, opts)
func (e *extendMsgDatabase) GetAllExtendMsgSet(ctx context.Context, sourceID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error) {
return e.db.GetAllExtendMsgSet(ctx, sourceID, opts)
}
func (e *ExtendMsgController) GetExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error) {
return e.GetExtendMsgSet(ctx, sourceID, sessionType, maxMsgUpdateTime)
func (e *extendMsgDatabase) GetExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error) {
return e.db.GetExtendMsgSet(ctx, sourceID, sessionType, maxMsgUpdateTime)
}
func (e *ExtendMsgController) InsertExtendMsg(ctx context.Context, sourceID string, sessionType int32, msg *unRelationTb.ExtendMsgModel) error {
return e.InsertExtendMsg(ctx, sourceID, sessionType, msg)
func (e *extendMsgDatabase) InsertExtendMsg(ctx context.Context, sourceID string, sessionType int32, msg *unRelationTb.ExtendMsgModel) error {
return e.db.InsertExtendMsg(ctx, sourceID, sessionType, msg)
}
func (e *ExtendMsgController) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error {
return e.InsertOrUpdateReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
func (e *extendMsgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error {
return e.db.InsertOrUpdateReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
}
func (e *ExtendMsgController) DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error {
return e.DeleteReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
func (e *extendMsgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error {
return e.db.DeleteReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
}
func (e *ExtendMsgController) GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error) {
return e.GetExtendMsg(ctx, sourceID, sessionType, clientMsgID, maxMsgUpdateTime)
}
type ExtendMsgDatabaseInterface interface {
CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error
GetAllExtendMsgSet(ctx context.Context, ID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error)
GetExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error)
InsertExtendMsg(ctx context.Context, sourceID string, sessionType int32, msg *unRelationTb.ExtendMsgModel) error
InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error)
}
type ExtendMsgDatabase struct {
model unRelationTb.ExtendMsgSetModelInterface
}
func NewExtendMsgDatabase() ExtendMsgDatabaseInterface {
return &ExtendMsgDatabase{}
}
func (e *ExtendMsgDatabase) CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error {
return e.model.CreateExtendMsgSet(ctx, set)
}
func (e *ExtendMsgDatabase) GetAllExtendMsgSet(ctx context.Context, sourceID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error) {
return e.model.GetAllExtendMsgSet(ctx, sourceID, opts)
}
func (e *ExtendMsgDatabase) GetExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error) {
return e.model.GetExtendMsgSet(ctx, sourceID, sessionType, maxMsgUpdateTime)
}
func (e *ExtendMsgDatabase) InsertExtendMsg(ctx context.Context, sourceID string, sessionType int32, msg *unRelationTb.ExtendMsgModel) error {
return e.model.InsertExtendMsg(ctx, sourceID, sessionType, msg)
}
func (e *ExtendMsgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error {
return e.model.InsertOrUpdateReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
}
func (e *ExtendMsgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error {
return e.model.DeleteReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
}
func (e *ExtendMsgDatabase) GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error) {
return e.model.TakeExtendMsg(ctx, sourceID, sessionType, clientMsgID, maxMsgUpdateTime)
func (e *extendMsgDatabase) GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error) {
return e.db.TakeExtendMsg(ctx, sourceID, sessionType, clientMsgID, maxMsgUpdateTime)
}
pkg/common/db/controller/group.go
+5 -12
View File
@@ -9,9 +9,7 @@ import (
"OpenIM/pkg/utils"
"context"
"fmt"
"github.com/dtm-labs/rockscache"
_ "github.com/dtm-labs/rockscache"
"github.com/go-redis/redis/v8"
)
type GroupDatabase interface {
@@ -56,8 +54,8 @@ func NewGroupDatabase(
request relationTb.GroupRequestModelInterface,
tx tx.Tx,
ctxTx tx.CtxTx,
super unRelationTb.SuperGroupModelInterface,
client redis.UniversalClient,
superGroup unRelationTb.SuperGroupModelInterface,
cache cache.GroupCache,
) GroupDatabase {
database := &groupDatabase{
groupDB: group,
@@ -65,13 +63,8 @@ func NewGroupDatabase(
groupRequestDB: request,
tx: tx,
ctxTx: ctxTx,
cache: cache.NewGroupCacheRedis(client, group, member, request, super, rockscache.Options{
RandomExpireAdjustment: 0.2,
DisableCacheRead: false,
DisableCacheDelete: false,
StrongConsistency: true,
}),
mongoDB: super,
cache: cache,
mongoDB: superGroup,
}
return database
}
@@ -82,7 +75,7 @@ type groupDatabase struct {
groupRequestDB relationTb.GroupRequestModelInterface
tx tx.Tx
ctxTx tx.CtxTx
cache cache.GroupCacheRedisInterface
cache cache.GroupCache
mongoDB unRelationTb.SuperGroupModelInterface
}
pkg/common/db/controller/msg.go
+5 -5
View File
@@ -552,8 +552,8 @@ func (db *msgDatabase) DeleteUserMsgsAndSetMinSeq(ctx context.Context, userID st
// this is struct for recursion
type delMsgRecursionStruct struct {
minSeq int64
delDocIDList []string
minSeq int64
delDocIDs []string
}
func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
@@ -576,7 +576,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
}
}
// 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList)
err = db.mgo.Delete(ctx, delStruct.delDocIDList)
err = db.mgo.Delete(ctx, delStruct.delDocIDs)
if err != nil {
return 0, err
}
@@ -587,7 +587,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
log.NewWarn(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), "msgs too large:", len(msgs.Msg), "docID:", msgs.DocID)
}
if msgs.Msg[len(msgs.Msg)-1].SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() && msgs.IsFull() {
delStruct.delDocIDList = append(delStruct.delDocIDList, msgs.DocID)
delStruct.delDocIDs = append(delStruct.delDocIDs, msgs.DocID)
lastMsgPb := &sdkws.MsgData{}
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
if err != nil {
@@ -611,7 +611,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
msg.SendTime = 0
hasMarkDelFlag = true
} else {
if err := db.mgo.Delete(ctx, delStruct.delDocIDList); err != nil {
if err := db.mgo.Delete(ctx, delStruct.delDocIDs); err != nil {
return 0, err
}
if hasMarkDelFlag {
pkg/common/db/controller/push.go
+7 -3
View File
@@ -5,14 +5,18 @@ import (
"context"
)
type PushInterface interface {
type PushDatabase interface {
DelFcmToken(ctx context.Context, userID string, platformID int) error
}
type PushDataBase struct {
type pushDataBase struct {
cache cache.Cache
}
func (p *PushDataBase) DelFcmToken(ctx context.Context, userID string, platformID int) error {
func NewPushDatabase(cache cache.Cache) PushDatabase {
return &pushDataBase{cache: cache}
}
func (p *pushDataBase) DelFcmToken(ctx context.Context, userID string, platformID int) error {
return p.cache.DelFcmToken(ctx, userID, platformID)
}
pkg/common/db/controller/third.go
+38
View File
@@ -0,0 +1,38 @@
package controller
import (
"OpenIM/pkg/common/db/cache"
"OpenIM/pkg/proto/sdkws"
"context"
)
type ThirdDatabase interface {
GetSignalInvitationInfoByClientMsgID(ctx context.Context, clientMsgID string) (invitationInfo *sdkws.SignalInviteReq, err error)
GetAvailableSignalInvitationInfo(ctx context.Context, userID string) (invitationInfo *sdkws.SignalInviteReq, err error)
FcmUpdateToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) error
SetAppBadge(ctx context.Context, userID string, value int) error
}
type thirdDatabase struct {
cache cache.Cache
}
func NewThirdDatabase(cache cache.Cache) ThirdDatabase {
return &thirdDatabase{cache: cache}
}
func (t *thirdDatabase) GetSignalInvitationInfoByClientMsgID(ctx context.Context, clientMsgID string) (invitationInfo *sdkws.SignalInviteReq, err error) {
return t.cache.GetSignalInvitationInfoByClientMsgID(ctx, clientMsgID)
}
func (t *thirdDatabase) GetAvailableSignalInvitationInfo(ctx context.Context, userID string) (invitationInfo *sdkws.SignalInviteReq, err error) {
return t.cache.GetAvailableSignalInvitationInfo(ctx, userID)
}
func (t *thirdDatabase) FcmUpdateToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) error {
return t.cache.SetFcmToken(ctx, account, platformID, fcmToken, expireTime)
}
func (t *thirdDatabase) SetAppBadge(ctx context.Context, userID string, value int) error {
return t.cache.SetUserBadgeUnreadCountSum(ctx, userID, value)
}