mirror of
https://github.com/openimsdk/open-im-server.git
synced 2025-04-27 20:30:40 +08:00
Merge remote-tracking branch 'origin/errcode' into errcode
This commit is contained in:
skiffer-git
commit
d59c4367fb
17
internal/common/check/group.go
Normal file
17
internal/common/check/group.go
Normal file
@ -0,0 +1,17 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
server_api_params "Open_IM/pkg/proto/sdk_ws"
|
||||
"errors"
|
||||
)
|
||||
|
||||
type GroupChecker struct {
|
||||
}
|
||||
|
||||
func NewGroupChecker() *GroupChecker {
|
||||
return &GroupChecker{}
|
||||
}
|
||||
|
||||
func (g *GroupChecker) GetGroupInfo(groupID string) (*server_api_params.GroupInfo, error) {
|
||||
return nil, errors.New("TODO:GetUserInfo")
|
||||
}
|
||||
@ -1,17 +1,22 @@
|
||||
package conversation
|
||||
|
||||
import (
|
||||
"Open_IM/internal/common/check"
|
||||
chat "Open_IM/internal/rpc/msg"
|
||||
"Open_IM/pkg/common/constant"
|
||||
"Open_IM/pkg/common/db"
|
||||
imdb "Open_IM/pkg/common/db/mysql_model/im_mysql_model"
|
||||
rocksCache "Open_IM/pkg/common/db/rocks_cache"
|
||||
"Open_IM/pkg/common/db/cache"
|
||||
"Open_IM/pkg/common/db/controller"
|
||||
"Open_IM/pkg/common/db/relation"
|
||||
"Open_IM/pkg/common/db/table"
|
||||
"Open_IM/pkg/common/db/unrelation"
|
||||
"Open_IM/pkg/common/log"
|
||||
promePkg "Open_IM/pkg/common/prometheus"
|
||||
"Open_IM/pkg/getcdv3"
|
||||
pbConversation "Open_IM/pkg/proto/conversation"
|
||||
pbUser "Open_IM/pkg/proto/user"
|
||||
"Open_IM/pkg/utils"
|
||||
"context"
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
@ -23,156 +28,55 @@ import (
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
type rpcConversation struct {
|
||||
type conversationServer struct {
|
||||
rpcPort int
|
||||
rpcRegisterName string
|
||||
etcdSchema string
|
||||
etcdAddr []string
|
||||
groupChecker *check.GroupChecker
|
||||
controller.ConversationInterface
|
||||
}
|
||||
|
||||
func (rpc *rpcConversation) ModifyConversationField(c context.Context, req *pbConversation.ModifyConversationFieldReq) (*pbConversation.ModifyConversationFieldResp, error) {
|
||||
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
|
||||
resp := &pbConversation.ModifyConversationFieldResp{}
|
||||
var err error
|
||||
isSyncConversation := true
|
||||
if req.Conversation.ConversationType == constant.GroupChatType {
|
||||
groupInfo, err := imdb.GetGroupInfoByGroupID(req.Conversation.GroupID)
|
||||
if err != nil {
|
||||
log.NewError(req.OperationID, "GetGroupInfoByGroupID failed ", req.Conversation.GroupID, err.Error())
|
||||
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
|
||||
return resp, nil
|
||||
}
|
||||
if groupInfo.Status == constant.GroupStatusDismissed && !req.Conversation.IsNotInGroup && req.FieldType != constant.FieldUnread {
|
||||
errMsg := "group status is dismissed"
|
||||
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: errMsg}
|
||||
return resp, nil
|
||||
}
|
||||
}
|
||||
var conversation imdb.Conversation
|
||||
if err := utils.CopyStructFields(&conversation, req.Conversation); err != nil {
|
||||
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields failed", *req.Conversation, err.Error())
|
||||
}
|
||||
haveUserID, _ := imdb.GetExistConversationUserIDList(req.UserIDList, req.Conversation.ConversationID)
|
||||
switch req.FieldType {
|
||||
case constant.FieldRecvMsgOpt:
|
||||
for _, v := range req.UserIDList {
|
||||
if err = db.DB.SetSingleConversationRecvMsgOpt(v, req.Conversation.ConversationID, req.Conversation.RecvMsgOpt); err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), "cache failed, rpc return", err.Error())
|
||||
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
|
||||
return resp, nil
|
||||
}
|
||||
}
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"recv_msg_opt": conversation.RecvMsgOpt})
|
||||
case constant.FieldGroupAtType:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"group_at_type": conversation.GroupAtType})
|
||||
case constant.FieldIsNotInGroup:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"is_not_in_group": conversation.IsNotInGroup})
|
||||
case constant.FieldIsPinned:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"is_pinned": conversation.IsPinned})
|
||||
case constant.FieldIsPrivateChat:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"is_private_chat": conversation.IsPrivateChat})
|
||||
case constant.FieldEx:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"ex": conversation.Ex})
|
||||
case constant.FieldAttachedInfo:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"attached_info": conversation.AttachedInfo})
|
||||
case constant.FieldUnread:
|
||||
isSyncConversation = false
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"update_unread_count_time": conversation.UpdateUnreadCountTime})
|
||||
case constant.FieldBurnDuration:
|
||||
err = imdb.UpdateColumnsConversations(haveUserID, req.Conversation.ConversationID, map[string]interface{}{"burn_duration": conversation.BurnDuration})
|
||||
}
|
||||
if err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), "UpdateColumnsConversations error", err.Error())
|
||||
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
|
||||
return resp, nil
|
||||
}
|
||||
for _, v := range utils.DifferenceString(haveUserID, req.UserIDList) {
|
||||
conversation.OwnerUserID = v
|
||||
err = rocksCache.DelUserConversationIDListFromCache(v)
|
||||
if err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), v, req.Conversation.ConversationID, err.Error())
|
||||
}
|
||||
err := imdb.SetOneConversation(conversation)
|
||||
if err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), "SetConversation error", err.Error())
|
||||
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
|
||||
return resp, nil
|
||||
}
|
||||
}
|
||||
|
||||
// notification
|
||||
if req.Conversation.ConversationType == constant.SingleChatType && req.FieldType == constant.FieldIsPrivateChat {
|
||||
//sync peer user conversation if conversation is singleChatType
|
||||
if err := syncPeerUserConversation(req.Conversation, req.OperationID); err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), "syncPeerUserConversation", err.Error())
|
||||
resp.CommonResp = &pbConversation.CommonResp{ErrCode: constant.ErrDB.ErrCode, ErrMsg: constant.ErrDB.ErrMsg}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
} else {
|
||||
if isSyncConversation {
|
||||
for _, v := range req.UserIDList {
|
||||
if err = rocksCache.DelConversationFromCache(v, req.Conversation.ConversationID); err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), v, req.Conversation.ConversationID, err.Error())
|
||||
}
|
||||
chat.ConversationChangeNotification(req.OperationID, v)
|
||||
}
|
||||
} else {
|
||||
for _, v := range req.UserIDList {
|
||||
if err = rocksCache.DelConversationFromCache(v, req.Conversation.ConversationID); err != nil {
|
||||
log.NewError(req.OperationID, utils.GetSelfFuncName(), v, req.Conversation.ConversationID, err.Error())
|
||||
}
|
||||
chat.ConversationUnreadChangeNotification(req.OperationID, v, req.Conversation.ConversationID, conversation.UpdateUnreadCountTime)
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "rpc return", resp.String())
|
||||
resp.CommonResp = &pbConversation.CommonResp{}
|
||||
return resp, nil
|
||||
}
|
||||
func syncPeerUserConversation(conversation *pbConversation.Conversation, operationID string) error {
|
||||
peerUserConversation := imdb.Conversation{
|
||||
OwnerUserID: conversation.UserID,
|
||||
ConversationID: utils.GetConversationIDBySessionType(conversation.OwnerUserID, constant.SingleChatType),
|
||||
ConversationType: constant.SingleChatType,
|
||||
UserID: conversation.OwnerUserID,
|
||||
GroupID: "",
|
||||
RecvMsgOpt: 0,
|
||||
UnreadCount: 0,
|
||||
DraftTextTime: 0,
|
||||
IsPinned: false,
|
||||
IsPrivateChat: conversation.IsPrivateChat,
|
||||
AttachedInfo: "",
|
||||
Ex: "",
|
||||
}
|
||||
err := imdb.PeerUserSetConversation(peerUserConversation)
|
||||
if err != nil {
|
||||
log.NewError(operationID, utils.GetSelfFuncName(), "SetConversation error", err.Error())
|
||||
return err
|
||||
}
|
||||
err = rocksCache.DelConversationFromCache(conversation.UserID, utils.GetConversationIDBySessionType(conversation.OwnerUserID, constant.SingleChatType))
|
||||
if err != nil {
|
||||
log.NewError(operationID, utils.GetSelfFuncName(), "DelConversationFromCache failed", err.Error(), conversation.OwnerUserID, conversation.ConversationID)
|
||||
}
|
||||
err = rocksCache.DelConversationFromCache(conversation.OwnerUserID, conversation.ConversationID)
|
||||
if err != nil {
|
||||
log.NewError(operationID, utils.GetSelfFuncName(), "DelConversationFromCache failed", err.Error(), conversation.OwnerUserID, conversation.ConversationID)
|
||||
}
|
||||
chat.ConversationSetPrivateNotification(operationID, conversation.OwnerUserID, conversation.UserID, conversation.IsPrivateChat)
|
||||
return nil
|
||||
}
|
||||
func NewRpcConversationServer(port int) *rpcConversation {
|
||||
func NewConversationServer(port int) *conversationServer {
|
||||
log.NewPrivateLog(constant.LogFileName)
|
||||
return &rpcConversation{
|
||||
c := conversationServer{
|
||||
rpcPort: port,
|
||||
rpcRegisterName: config.Config.RpcRegisterName.OpenImConversationName,
|
||||
etcdSchema: config.Config.Etcd.EtcdSchema,
|
||||
etcdAddr: config.Config.Etcd.EtcdAddr,
|
||||
groupChecker: check.NewGroupChecker(),
|
||||
}
|
||||
var cDB relation.Conversation
|
||||
var cCache cache.ConversationCache
|
||||
//mysql init
|
||||
var mysql relation.Mysql
|
||||
err := mysql.InitConn().AutoMigrateModel(&table.ConversationModel{})
|
||||
if err != nil {
|
||||
panic("db init err:" + err.Error())
|
||||
}
|
||||
if mysql.GormConn() != nil {
|
||||
//get gorm model
|
||||
cDB = relation.NewConversationGorm(mysql.GormConn())
|
||||
} else {
|
||||
panic("db init err:" + "conn is nil")
|
||||
}
|
||||
//redis init
|
||||
var redis cache.RedisClient
|
||||
redis.InitRedis()
|
||||
rcClient := rockscache.NewClient(redis.GetClient(), rockscache.Options{
|
||||
RandomExpireAdjustment: 0.2,
|
||||
DisableCacheRead: false,
|
||||
DisableCacheDelete: false,
|
||||
StrongConsistency: true,
|
||||
})
|
||||
cCache = cache.NewConversationRedis(rcClient)
|
||||
|
||||
database := controller.NewConversationDataBase(cDB, cCache)
|
||||
c.ConversationInterface = controller.NewConversationController(database)
|
||||
return &c
|
||||
}
|
||||
|
||||
func (rpc *rpcConversation) Run() {
|
||||
func (c *conversationServer) Run() {
|
||||
log.NewInfo("0", "rpc conversation start...")
|
||||
|
||||
listenIP := ""
|
||||
@ -181,11 +85,11 @@ func (rpc *rpcConversation) Run() {
|
||||
} else {
|
||||
listenIP = config.Config.ListenIP
|
||||
}
|
||||
address := listenIP + ":" + strconv.Itoa(rpc.rpcPort)
|
||||
address := listenIP + ":" + strconv.Itoa(c.rpcPort)
|
||||
|
||||
listener, err := net.Listen("tcp", address)
|
||||
if err != nil {
|
||||
panic("listening err:" + err.Error() + rpc.rpcRegisterName)
|
||||
panic("listening err:" + err.Error() + c.rpcRegisterName)
|
||||
}
|
||||
log.NewInfo("0", "listen network success, ", address, listener)
|
||||
//grpc server
|
||||
@ -204,7 +108,7 @@ func (rpc *rpcConversation) Run() {
|
||||
defer srv.GracefulStop()
|
||||
|
||||
//service registers with etcd
|
||||
pbConversation.RegisterConversationServer(srv, rpc)
|
||||
pbConversation.RegisterConversationServer(srv, c)
|
||||
rpcRegisterIP := config.Config.RpcRegisterIP
|
||||
if config.Config.RpcRegisterIP == "" {
|
||||
rpcRegisterIP, err = utils.GetLocalIP()
|
||||
@ -213,13 +117,13 @@ func (rpc *rpcConversation) Run() {
|
||||
}
|
||||
}
|
||||
log.NewInfo("", "rpcRegisterIP", rpcRegisterIP)
|
||||
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName, 10)
|
||||
err = getcdv3.RegisterEtcd(c.etcdSchema, strings.Join(c.etcdAddr, ","), rpcRegisterIP, c.rpcPort, c.rpcRegisterName, 10, "")
|
||||
if err != nil {
|
||||
log.NewError("0", "RegisterEtcd failed ", err.Error(),
|
||||
rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName)
|
||||
c.etcdSchema, strings.Join(c.etcdAddr, ","), rpcRegisterIP, c.rpcPort, c.rpcRegisterName)
|
||||
panic(utils.Wrap(err, "register conversation module rpc to etcd err"))
|
||||
}
|
||||
log.NewInfo("0", "RegisterConversationServer ok ", rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName)
|
||||
log.NewInfo("0", "RegisterConversationServer ok ", c.etcdSchema, strings.Join(c.etcdAddr, ","), rpcRegisterIP, c.rpcPort, c.rpcRegisterName)
|
||||
err = srv.Serve(listener)
|
||||
if err != nil {
|
||||
log.NewError("0", "Serve failed ", err.Error())
|
||||
@ -227,3 +131,141 @@ func (rpc *rpcConversation) Run() {
|
||||
}
|
||||
log.NewInfo("0", "rpc conversation ok")
|
||||
}
|
||||
|
||||
func (c *conversationServer) GetConversation(ctx context.Context, req *pbConversation.GetConversationReq) (*pbConversation.GetConversationResp, error) {
|
||||
resp := &pbConversation.GetConversationResp{Conversation: &pbConversation.Conversation{}}
|
||||
conversations, err := c.ConversationInterface.FindConversations(ctx, req.OwnerUserID, []string{req.ConversationID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(conversations) > 0 {
|
||||
if err := utils.CopyStructFields(resp.Conversation, &conversations[0]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (c *conversationServer) GetAllConversations(ctx context.Context, req *pbConversation.GetAllConversationsReq) (*pbConversation.GetAllConversationsResp, error) {
|
||||
resp := &pbConversation.GetAllConversationsResp{Conversations: []*pbConversation.Conversation{}}
|
||||
conversations, err := c.ConversationInterface.GetUserAllConversation(ctx, req.OwnerUserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := utils.CopyStructFields(&resp.Conversations, conversations); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (c *conversationServer) GetConversations(ctx context.Context, req *pbConversation.GetConversationsReq) (*pbConversation.GetConversationsResp, error) {
|
||||
resp := &pbConversation.GetConversationsResp{Conversations: []*pbConversation.Conversation{}}
|
||||
conversations, err := c.ConversationInterface.FindConversations(ctx, req.OwnerUserID, req.ConversationIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := utils.CopyStructFields(&resp.Conversations, conversations); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (c *conversationServer) BatchSetConversations(ctx context.Context, req *pbConversation.BatchSetConversationsReq) (*pbConversation.BatchSetConversationsResp, error) {
|
||||
resp := &pbConversation.BatchSetConversationsResp{}
|
||||
var conversations []*table.ConversationModel
|
||||
if err := utils.CopyStructFields(&conversations, req.Conversations); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
err := c.ConversationInterface.SetUserConversations(ctx, req.OwnerUserID, conversations)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
chat.ConversationChangeNotification(ctx, req.OwnerUserID)
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (c *conversationServer) SetConversation(ctx context.Context, req *pbConversation.SetConversationReq) (*pbConversation.SetConversationResp, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *conversationServer) SetRecvMsgOpt(ctx context.Context, req *pbConversation.SetRecvMsgOptReq) (*pbConversation.SetRecvMsgOptResp, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *conversationServer) ModifyConversationField(ctx context.Context, req *pbConversation.ModifyConversationFieldReq) (*pbConversation.ModifyConversationFieldResp, error) {
|
||||
resp := &pbConversation.ModifyConversationFieldResp{}
|
||||
var err error
|
||||
isSyncConversation := true
|
||||
if req.Conversation.ConversationType == constant.GroupChatType {
|
||||
groupInfo, err := c.groupChecker.GetGroupInfo(req.Conversation.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if groupInfo.Status == constant.GroupStatusDismissed && req.FieldType != constant.FieldUnread {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
var conversation table.ConversationModel
|
||||
if err := utils.CopyStructFields(&conversation, req.Conversation); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if req.FieldType == constant.FieldIsPrivateChat {
|
||||
err := c.ConversationInterface.SyncPeerUserPrivateConversationTx(ctx, req.Conversation)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
chat.ConversationSetPrivateNotification(req.OperationID, req.Conversation.OwnerUserID, req.Conversation.UserID, req.Conversation.IsPrivateChat)
|
||||
return resp, nil
|
||||
}
|
||||
//haveUserID, err := c.ConversationInterface.GetUserIDExistConversation(ctx, req.UserIDList, req.Conversation.ConversationID)
|
||||
//if err != nil {
|
||||
// return nil, err
|
||||
//}
|
||||
filedMap := make(map[string]interface{})
|
||||
switch req.FieldType {
|
||||
case constant.FieldRecvMsgOpt:
|
||||
filedMap["recv_msg_opt"] = req.Conversation.RecvMsgOpt
|
||||
case constant.FieldGroupAtType:
|
||||
filedMap["group_at_type"] = req.Conversation.GroupAtType
|
||||
case constant.FieldIsNotInGroup:
|
||||
filedMap["is_not_in_group"] = req.Conversation.IsNotInGroup
|
||||
case constant.FieldIsPinned:
|
||||
filedMap["is_pinned"] = req.Conversation.IsPinned
|
||||
case constant.FieldEx:
|
||||
filedMap["ex"] = req.Conversation.Ex
|
||||
case constant.FieldAttachedInfo:
|
||||
filedMap["attached_info"] = req.Conversation.AttachedInfo
|
||||
case constant.FieldUnread:
|
||||
isSyncConversation = false
|
||||
filedMap["update_unread_count_time"] = req.Conversation.UpdateUnreadCountTime
|
||||
case constant.FieldBurnDuration:
|
||||
filedMap["burn_duration"] = req.Conversation.BurnDuration
|
||||
}
|
||||
c.ConversationInterface.SetUsersConversationFiledTx(ctx, req.UserIDList, &conversation, filedMap)
|
||||
err = c.ConversationInterface.UpdateUsersConversationFiled(ctx, haveUserID, req.Conversation.ConversationID, filedMap)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var conversations []*pbConversation.Conversation
|
||||
for _, v := range utils.DifferenceString(haveUserID, req.UserIDList) {
|
||||
temp := new(pbConversation.Conversation)
|
||||
_ = utils.CopyStructFields(temp, req.Conversation)
|
||||
temp.OwnerUserID = v
|
||||
conversations = append(conversations, temp)
|
||||
}
|
||||
err = c.ConversationInterface.CreateConversation(ctx, conversations)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if isSyncConversation {
|
||||
for _, v := range req.UserIDList {
|
||||
chat.ConversationChangeNotification(req.OperationID, v)
|
||||
}
|
||||
} else {
|
||||
for _, v := range req.UserIDList {
|
||||
chat.ConversationUnreadChangeNotification(req.OperationID, v, req.Conversation.ConversationID, req.Conversation.UpdateUnreadCountTime)
|
||||
}
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
@ -6,6 +6,7 @@ import (
|
||||
"Open_IM/pkg/common/log"
|
||||
open_im_sdk "Open_IM/pkg/proto/sdk_ws"
|
||||
"Open_IM/pkg/utils"
|
||||
"context"
|
||||
"github.com/golang/protobuf/jsonpb"
|
||||
"github.com/golang/protobuf/proto"
|
||||
)
|
||||
@ -59,7 +60,7 @@ func ConversationSetPrivateNotification(operationID, sendID, recvID string, isPr
|
||||
}
|
||||
|
||||
// 会话改变
|
||||
func ConversationChangeNotification(operationID, userID string) {
|
||||
func ConversationChangeNotification(ctx context.Context, userID string) {
|
||||
log.NewInfo(operationID, utils.GetSelfFuncName())
|
||||
ConversationChangedTips := &open_im_sdk.ConversationUpdateTips{
|
||||
UserID: userID,
|
||||
|
||||
269
pkg/common/db/cache/conversation.go
vendored
269
pkg/common/db/cache/conversation.go
vendored
@ -1,253 +1,86 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/db/relation"
|
||||
relation2 "Open_IM/pkg/common/db/table/relation"
|
||||
"Open_IM/pkg/common/tracelog"
|
||||
"Open_IM/pkg/common/db/table"
|
||||
"Open_IM/pkg/utils"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"github.com/go-redis/redis/v8"
|
||||
"golang.org/x/tools/go/ssa/testdata/src/strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
conversationKey = "CONVERSATION:"
|
||||
conversationIDsKey = "CONVERSATION_IDS:"
|
||||
recvMsgOptKey = "RECV_MSG_OPT:"
|
||||
superGroupRecvMsgNotNotifyUserIDsKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
|
||||
conversationExpireTime = time.Second * 60 * 60 * 12
|
||||
)
|
||||
type DBFun func() (string, error)
|
||||
|
||||
type ConversationCache struct {
|
||||
conversationDB *relation.ConversationGorm
|
||||
expireTime time.Duration
|
||||
rcClient *rockscache.Client
|
||||
type ConversationCache interface {
|
||||
GetUserConversationIDListFromCache(userID string, fn DBFun) ([]string, error)
|
||||
DelUserConversationIDListFromCache(userID string) error
|
||||
GetConversationFromCache(ownerUserID, conversationID string, fn DBFun) (*table.ConversationModel, error)
|
||||
GetConversationsFromCache(ownerUserID string, conversationIDList []string, fn DBFun) ([]*table.ConversationModel, error)
|
||||
GetUserAllConversationList(ownerUserID string, fn DBFun) ([]*table.ConversationModel, error)
|
||||
DelConversationFromCache(ownerUserID, conversationID string) error
|
||||
}
|
||||
type ConversationRedis struct {
|
||||
rcClient *rockscache.Client
|
||||
}
|
||||
|
||||
func NewConversationCache(rdb redis.UniversalClient, conversationDB *relation.ConversationGorm, options rockscache.Options) *ConversationCache {
|
||||
return &ConversationCache{conversationDB: conversationDB, expireTime: conversationExpireTime, rcClient: rockscache.NewClient(rdb, options)}
|
||||
func NewConversationRedis(rcClient *rockscache.Client) *ConversationRedis {
|
||||
return &ConversationRedis{rcClient: rcClient}
|
||||
}
|
||||
|
||||
func (c *ConversationCache) getConversationKey(ownerUserID, conversationID string) string {
|
||||
return conversationKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationCache) getConversationIDsKey(ownerUserID string) string {
|
||||
return conversationIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
func (c *ConversationCache) getRecvMsgOptKey(ownerUserID, conversationID string) string {
|
||||
return recvMsgOptKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationCache) getSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
|
||||
return superGroupRecvMsgNotNotifyUserIDsKey + groupID
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) (conversationIDs []string, err error) {
|
||||
//getConversationIDs := func() (string, error) {
|
||||
// conversationIDs, err := relation.GetConversationIDsByUserID(ownerUserID)
|
||||
// if err != nil {
|
||||
// return "", err
|
||||
// }
|
||||
// bytes, err := json.Marshal(conversationIDs)
|
||||
// if err != nil {
|
||||
// return "", utils.Wrap(err, "")
|
||||
// }
|
||||
// return string(bytes), nil
|
||||
//}
|
||||
//defer func() {
|
||||
// tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversationIDs", conversationIDs)
|
||||
//}()
|
||||
//conversationIDsStr, err := c.rcClient.Fetch(c.getConversationIDsKey(ownerUserID), time.Second*30*60, getConversationIDs)
|
||||
//err = json.Unmarshal([]byte(conversationIDsStr), &conversationIDs)
|
||||
//if err != nil {
|
||||
// return nil, utils.Wrap(err, "")
|
||||
//}
|
||||
//return conversationIDs, nil
|
||||
return GetCache(c.rcClient, c.getConversationIDsKey(ownerUserID), time.Second*30*60, func() ([]string, error) {
|
||||
return relation.GetConversationIDsByUserID(ownerUserID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetUserConversationIDs1(ctx context.Context, ownerUserID string, fn func() (any, error)) (conversationIDs []string, err error) {
|
||||
//getConversationIDs := func() (string, error) {
|
||||
// conversationIDs, err := relation.GetConversationIDsByUserID(ownerUserID)
|
||||
// if err != nil {
|
||||
// return "", err
|
||||
// }
|
||||
// bytes, err := json.Marshal(conversationIDs)
|
||||
// if err != nil {
|
||||
// return "", utils.Wrap(err, "")
|
||||
// }
|
||||
// return string(bytes), nil
|
||||
//}
|
||||
//defer func() {
|
||||
// tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversationIDs", conversationIDs)
|
||||
//}()
|
||||
//conversationIDsStr, err := c.rcClient.Fetch(c.getConversationIDsKey(ownerUserID), time.Second*30*60, getConversationIDs)
|
||||
//err = json.Unmarshal([]byte(conversationIDsStr), &conversationIDs)
|
||||
//if err != nil {
|
||||
// return nil, utils.Wrap(err, "")
|
||||
//}
|
||||
//return conversationIDs, nil
|
||||
return GetCache1[[]string](c.rcClient, c.getConversationIDsKey(ownerUserID), time.Second*30*60, fn)
|
||||
}
|
||||
|
||||
func GetCache1[T any](rcClient *rockscache.Client, key string, expire time.Duration, fn func() (any, error)) (T, error) {
|
||||
v, err := rcClient.Fetch(key, expire, func() (string, error) {
|
||||
v, err := fn()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
bs, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return "", utils.Wrap(err, "")
|
||||
}
|
||||
return string(bs), nil
|
||||
})
|
||||
var t T
|
||||
func (c *ConversationRedis) GetUserConversationIDListFromCache(userID string, fn DBFun) ([]string, error) {
|
||||
conversationIDListStr, err := c.rcClient.Fetch(conversationIDListCache+userID, time.Second*30*60, fn)
|
||||
var conversationIDList []string
|
||||
err = json.Unmarshal([]byte(conversationIDListStr), &conversationIDList)
|
||||
if err != nil {
|
||||
return t, err
|
||||
return nil, utils.Wrap(err, "")
|
||||
}
|
||||
err = json.Unmarshal([]byte(v), &t)
|
||||
if err != nil {
|
||||
return t, utils.Wrap(err, "")
|
||||
}
|
||||
return t, nil
|
||||
return conversationIDList, nil
|
||||
}
|
||||
|
||||
func GetCache[T any](rcClient *rockscache.Client, key string, expire time.Duration, fn func() (T, error)) (T, error) {
|
||||
v, err := rcClient.Fetch(key, expire, func() (string, error) {
|
||||
v, err := fn()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
bs, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return "", utils.Wrap(err, "")
|
||||
}
|
||||
return string(bs), nil
|
||||
})
|
||||
var t T
|
||||
if err != nil {
|
||||
return t, err
|
||||
}
|
||||
err = json.Unmarshal([]byte(v), &t)
|
||||
if err != nil {
|
||||
return t, utils.Wrap(err, "")
|
||||
}
|
||||
return t, nil
|
||||
func (c *ConversationRedis) DelUserConversationIDListFromCache(userID string) error {
|
||||
return utils.Wrap(c.rcClient.TagAsDeleted(conversationIDListCache+userID), "DelUserConversationIDListFromCache err")
|
||||
}
|
||||
|
||||
func (c *ConversationCache) DelUserConversationIDs(ctx context.Context, ownerUserID string) (err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID)
|
||||
}()
|
||||
return utils.Wrap(c.rcClient.TagAsDeleted(c.getConversationIDsKey(ownerUserID)), "DelUserConversationIDs err")
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetConversation(ctx context.Context, ownerUserID, conversationID string) (conversation *relation2.ConversationModel, err error) {
|
||||
getConversation := func() (string, error) {
|
||||
conversation, err := relation.GetConversation(ownerUserID, conversationID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
bytes, err := json.Marshal(conversation)
|
||||
if err != nil {
|
||||
return "", utils.Wrap(err, "conversation Marshal failed")
|
||||
}
|
||||
return string(bytes), nil
|
||||
}
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversationID", conversationID, "conversation", *conversation)
|
||||
}()
|
||||
conversationStr, err := c.rcClient.Fetch(c.getConversationKey(ownerUserID, conversationID), c.expireTime, getConversation)
|
||||
func (c *ConversationRedis) GetConversationFromCache(ownerUserID, conversationID string, fn DBFun) (*table.ConversationModel, error) {
|
||||
conversationStr, err := c.rcClient.Fetch(conversationCache+ownerUserID+":"+conversationID, time.Second*30*60, fn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, utils.Wrap(err, "Fetch failed")
|
||||
}
|
||||
conversation = &relation2.ConversationModel{}
|
||||
conversation := table.ConversationModel{}
|
||||
err = json.Unmarshal([]byte(conversationStr), &conversation)
|
||||
return conversation, utils.Wrap(err, "Unmarshal failed")
|
||||
}
|
||||
|
||||
func (c *ConversationCache) DelConversation(ctx context.Context, ownerUserID, conversationID string) (err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversationID", conversationID)
|
||||
}()
|
||||
return utils.Wrap(c.rcClient.TagAsDeleted(c.getConversationKey(ownerUserID, conversationID)), "DelConversation err")
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetConversations(ctx context.Context, ownerUserID string, conversationIDs []string) (conversations []relation2.ConversationModel, err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversationIDs", conversationIDs, "conversations", conversations)
|
||||
}()
|
||||
for _, conversationID := range conversationIDs {
|
||||
conversation, err := c.GetConversation(ctx, ownerUserID, conversationID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conversations = append(conversations, *conversation)
|
||||
if err != nil {
|
||||
return nil, utils.Wrap(err, "Unmarshal failed")
|
||||
}
|
||||
return conversations, nil
|
||||
return &conversation, nil
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetUserAllConversations(ctx context.Context, ownerUserID string) (conversations []relation2.ConversationModel, err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversations", conversations)
|
||||
}()
|
||||
IDs, err := c.GetUserConversationIDs(ctx, ownerUserID)
|
||||
func (c *ConversationRedis) GetConversationsFromCache(ownerUserID string, conversationIDList []string, fn DBFun) ([]*table.ConversationModel, error) {
|
||||
var conversationList []*table.ConversationModel
|
||||
for _, conversationID := range conversationIDList {
|
||||
conversation, err := c.GetConversationFromCache(ownerUserID, conversationID, fn)
|
||||
if err != nil {
|
||||
return nil, utils.Wrap(err, "GetConversationFromCache failed")
|
||||
}
|
||||
conversationList = append(conversationList, conversation)
|
||||
}
|
||||
return conversationList, nil
|
||||
}
|
||||
|
||||
func (c *ConversationRedis) GetUserAllConversationList(ownerUserID string, fn DBFun) ([]*table.ConversationModel, error) {
|
||||
IDList, err := c.GetUserConversationIDListFromCache(ownerUserID, fn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var conversationIDs []relation2.ConversationModel
|
||||
for _, conversationID := range IDs {
|
||||
conversation, err := c.GetConversation(ctx, ownerUserID, conversationID)
|
||||
var conversationList []*table.ConversationModel
|
||||
for _, conversationID := range IDList {
|
||||
conversation, err := c.GetConversationFromCache(ownerUserID, conversationID, fn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, utils.Wrap(err, "GetConversationFromCache failed")
|
||||
}
|
||||
conversationIDs = append(conversationIDs, *conversation)
|
||||
conversationList = append(conversationList, conversation)
|
||||
}
|
||||
return conversationIDs, nil
|
||||
return conversationList, nil
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error) {
|
||||
getConversation := func() (string, error) {
|
||||
conversation, err := relation.GetConversation(ownerUserID, conversationID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return strconv.Itoa(int(conversation.RecvMsgOpt)), nil
|
||||
}
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "ownerUserID", ownerUserID, "conversationID", conversationID, "opt", opt)
|
||||
}()
|
||||
optStr, err := c.rcClient.Fetch(c.getConversationKey(ownerUserID, conversationID), c.expireTime, getConversation)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return strconv.Atoi(optStr)
|
||||
}
|
||||
|
||||
func (c *ConversationCache) DelUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) error {
|
||||
return utils.Wrap(c.rcClient.TagAsDeleted(c.getConversationKey(ownerUserID, conversationID)), "DelUserRecvMsgOpt failed")
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetSuperGroupRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) (userIDs []string, err error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (c *ConversationCache) DelSuperGroupRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ConversationCache) GetSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) (hash uint32, err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (c *ConversationCache) DelSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) {
|
||||
return
|
||||
func (c *ConversationRedis) DelConversationFromCache(ownerUserID, conversationID string) error {
|
||||
return utils.Wrap(c.rcClient.TagAsDeleted(conversationCache+ownerUserID+":"+conversationID), "DelConversationFromCache err")
|
||||
}
|
||||
|
||||
119
pkg/common/db/controller/conversation.go
Normal file
119
pkg/common/db/controller/conversation.go
Normal file
@ -0,0 +1,119 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/db/cache"
|
||||
"Open_IM/pkg/common/db/relation"
|
||||
"Open_IM/pkg/common/db/table"
|
||||
"context"
|
||||
)
|
||||
|
||||
type ConversationInterface interface {
|
||||
//GetUserIDExistConversation 获取拥有该会话的的用户ID列表
|
||||
GetUserIDExistConversation(ctx context.Context, userIDList []string, conversationID string) ([]string, error)
|
||||
//UpdateUserConversationFiled 更新用户该会话的属性信息
|
||||
UpdateUsersConversationFiled(ctx context.Context, UserIDList []string, conversationID string, args map[string]interface{}) error
|
||||
//CreateConversation 创建一批新的会话
|
||||
CreateConversation(ctx context.Context, conversations []*table.ConversationModel) error
|
||||
//SyncPeerUserPrivateConversation 同步对端私聊会话内部保证事务操作
|
||||
SyncPeerUserPrivateConversationTx(ctx context.Context, conversation *table.ConversationModel) error
|
||||
//FindConversations 根据会话ID获取某个用户的多个会话
|
||||
FindConversations(ctx context.Context, ownerUserID string, conversationID []string) ([]*table.ConversationModel, error)
|
||||
//GetUserAllConversation 获取一个用户在服务器上所有的会话
|
||||
GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*table.ConversationModel, error)
|
||||
//SetUserConversations 设置用户多个会话属性，如果会话不存在则创建，否则更新,内部保证原子性
|
||||
SetUserConversations(ctx context.Context, ownerUserID string, conversations []*table.ConversationModel) error
|
||||
}
|
||||
type ConversationController struct {
|
||||
database ConversationDataBaseInterface
|
||||
}
|
||||
|
||||
func NewConversationController(database ConversationDataBaseInterface) *ConversationController {
|
||||
return &ConversationController{database: database}
|
||||
}
|
||||
|
||||
func (c *ConversationController) GetUserIDExistConversation(ctx context.Context, userIDList []string, conversationID string) ([]string, error) {
|
||||
return c.database.GetUserIDExistConversation(ctx, userIDList, conversationID)
|
||||
}
|
||||
|
||||
func (c ConversationController) UpdateUsersConversationFiled(ctx context.Context, UserIDList []string, conversationID string, args map[string]interface{}) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationController) CreateConversation(ctx context.Context, conversations []*table.ConversationModel) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationController) SyncPeerUserPrivateConversationTx(ctx context.Context, conversation *table.ConversationModel) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationController) FindConversations(ctx context.Context, ownerUserID string, conversationID []string) ([]*table.ConversationModel, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationController) GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*table.ConversationModel, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
func (c ConversationController) SetUserConversations(ctx context.Context, ownerUserID string, conversations []*table.ConversationModel) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
var _ ConversationInterface = (*ConversationController)(nil)
|
||||
|
||||
type ConversationDataBaseInterface interface {
|
||||
//GetUserIDExistConversation 获取拥有该会话的的用户ID列表
|
||||
GetUserIDExistConversation(ctx context.Context, userIDList []string, conversationID string) ([]string, error)
|
||||
//UpdateUserConversationFiled 更新用户该会话的属性信息
|
||||
UpdateUsersConversationFiled(ctx context.Context, UserIDList []string, conversationID string, args map[string]interface{}) error
|
||||
//CreateConversation 创建一批新的会话
|
||||
CreateConversation(ctx context.Context, conversations []*table.ConversationModel) error
|
||||
//SyncPeerUserPrivateConversation 同步对端私聊会话内部保证事务操作
|
||||
SyncPeerUserPrivateConversationTx(ctx context.Context, conversation *table.ConversationModel) error
|
||||
//FindConversations 根据会话ID获取某个用户的多个会话
|
||||
FindConversations(ctx context.Context, ownerUserID string, conversationID []string) ([]*table.ConversationModel, error)
|
||||
//GetUserAllConversation 获取一个用户在服务器上所有的会话
|
||||
GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*table.ConversationModel, error)
|
||||
//SetUserConversations 设置用户多个会话属性，如果会话不存在则创建，否则更新,内部保证原子性
|
||||
SetUserConversations(ctx context.Context, ownerUserID string, conversations []*table.ConversationModel) error
|
||||
}
|
||||
type ConversationDataBase struct {
|
||||
db relation.Conversation
|
||||
cache cache.ConversationCache
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) GetUserIDExistConversation(ctx context.Context, userIDList []string, conversationID string) ([]string, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) UpdateUsersConversationFiled(ctx context.Context, UserIDList []string, conversationID string, args map[string]interface{}) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) CreateConversation(ctx context.Context, conversations []*table.ConversationModel) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) SyncPeerUserPrivateConversationTx(ctx context.Context, conversation *table.ConversationModel) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) FindConversations(ctx context.Context, ownerUserID string, conversationID []string) ([]*table.ConversationModel, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*table.ConversationModel, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c ConversationDataBase) SetUserConversations(ctx context.Context, ownerUserID string, conversations []*table.ConversationModel) error {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func NewConversationDataBase(db relation.Conversation, cache cache.ConversationCache) *ConversationDataBase {
|
||||
return &ConversationDataBase{db: db, cache: cache}
|
||||
}
|
||||
|
||||
//func NewConversationController(db *gorm.DB, rdb redis.UniversalClient) ConversationInterface {
|
||||
// groupController := &ConversationController{database: newGroupDatabase(db, rdb, mgoClient)}
|
||||
// return groupController
|
||||
//}
|
||||
@ -1,15 +1,35 @@
|
||||
package relation
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/db/table/relation"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type ConversationGorm struct {
|
||||
DB *gorm.DB
|
||||
var ConversationDB *gorm.DB
|
||||
|
||||
//type Conversation struct {
|
||||
// OwnerUserID string `gorm:"column:owner_user_id;primary_key;type:char(128)" json:"OwnerUserID"`
|
||||
// ConversationID string `gorm:"column:conversation_id;primary_key;type:char(128)" json:"conversationID"`
|
||||
// ConversationType int32 `gorm:"column:conversation_type" json:"conversationType"`
|
||||
// UserID string `gorm:"column:user_id;type:char(64)" json:"userID"`
|
||||
// GroupID string `gorm:"column:group_id;type:char(128)" json:"groupID"`
|
||||
// RecvMsgOpt int32 `gorm:"column:recv_msg_opt" json:"recvMsgOpt"`
|
||||
// UnreadCount int32 `gorm:"column:unread_count" json:"unreadCount"`
|
||||
// DraftTextTime int64 `gorm:"column:draft_text_time" json:"draftTextTime"`
|
||||
// IsPinned bool `gorm:"column:is_pinned" json:"isPinned"`
|
||||
// IsPrivateChat bool `gorm:"column:is_private_chat" json:"isPrivateChat"`
|
||||
// BurnDuration int32 `gorm:"column:burn_duration;default:30" json:"burnDuration"`
|
||||
// GroupAtType int32 `gorm:"column:group_at_type" json:"groupAtType"`
|
||||
// IsNotInGroup bool `gorm:"column:is_not_in_group" json:"isNotInGroup"`
|
||||
// UpdateUnreadCountTime int64 `gorm:"column:update_unread_count_time" json:"updateUnreadCountTime"`
|
||||
// AttachedInfo string `gorm:"column:attached_info;type:varchar(1024)" json:"attachedInfo"`
|
||||
// Ex string `gorm:"column:ex;type:varchar(1024)" json:"ex"`
|
||||
//}
|
||||
|
||||
func (Conversation) TableName() string {
|
||||
return "conversations"
|
||||
}
|
||||
|
||||
func SetConversation(conversation relation.ConversationModel) (bool, error) {
|
||||
func SetConversation(conversation Conversation) (bool, error) {
|
||||
var isUpdate bool
|
||||
newConversation := conversation
|
||||
if ConversationDB.Model(&Conversation{}).Find(&newConversation).RowsAffected == 0 {
|
||||
@ -73,7 +93,7 @@ func GetExistConversationUserIDList(ownerUserIDList []string, conversationID str
|
||||
return resultArr, nil
|
||||
}
|
||||
|
||||
func GetConversation(OwnerUserID, conversationID string) (relation.ConversationModel, error) {
|
||||
func GetConversation(OwnerUserID, conversationID string) (Conversation, error) {
|
||||
var conversation Conversation
|
||||
err := ConversationDB.Table("conversations").Where("owner_user_id=? and conversation_id=?", OwnerUserID, conversationID).Take(&conversation).Error
|
||||
return conversation, err
|
||||
@ -96,7 +116,7 @@ func UpdateColumnsConversations(ownerUserIDList []string, conversationID string,
|
||||
|
||||
}
|
||||
|
||||
func GetConversationIDsByUserID(userID string) ([]string, error) {
|
||||
func GetConversationIDListByUserID(userID string) ([]string, error) {
|
||||
var IDList []string
|
||||
err := ConversationDB.Model(&Conversation{}).Where("owner_user_id=?", userID).Pluck("conversation_id", &IDList).Error
|
||||
return IDList, err
|
||||
|
||||
73
pkg/common/db/relation/conversation_model_g.go
Normal file
73
pkg/common/db/relation/conversation_model_g.go
Normal file
@ -0,0 +1,73 @@
|
||||
package relation
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/db/table"
|
||||
"Open_IM/pkg/common/tracelog"
|
||||
"Open_IM/pkg/utils"
|
||||
"context"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type Conversation interface {
|
||||
TableName() string
|
||||
Create(ctx context.Context, conversations []*table.ConversationModel) (err error)
|
||||
Delete(ctx context.Context, groupIDs []string) (err error)
|
||||
UpdateByMap(ctx context.Context, groupID string, args map[string]interface{}) (err error)
|
||||
Update(ctx context.Context, groups []*table.ConversationModel) (err error)
|
||||
Find(ctx context.Context, groupIDs []string) (groups []*table.ConversationModel, err error)
|
||||
Take(ctx context.Context, groupID string) (group *table.ConversationModel, err error)
|
||||
}
|
||||
type ConversationGorm struct {
|
||||
DB *gorm.DB
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) TableName() string {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func NewConversationGorm(DB *gorm.DB) Conversation {
|
||||
return &ConversationGorm{DB: DB}
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) Create(ctx context.Context, conversations []*table.ConversationModel) (err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "conversations", conversations)
|
||||
}()
|
||||
return utils.Wrap(getDBConn(g.DB, tx).Create(&conversations).Error, "")
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) Delete(ctx context.Context, groupIDs []string) (err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupIDs", groupIDs)
|
||||
}()
|
||||
return utils.Wrap(getDBConn(g.DB, tx).Where("group_id in (?)", groupIDs).Delete(&table.ConversationModel{}).Error, "")
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) UpdateByMap(ctx context.Context, groupID string, args map[string]interface{}) (err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupID", groupID, "args", args)
|
||||
}()
|
||||
return utils.Wrap(getDBConn(g.DB, tx).Where("group_id = ?", groupID).Model(g).Updates(args).Error, "")
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) Update(ctx context.Context, groups []*table.ConversationModel) (err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groups", groups)
|
||||
}()
|
||||
return utils.Wrap(getDBConn(g.DB, tx).Updates(&groups).Error, "")
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) Find(ctx context.Context, groupIDs []string) (groups []*table.ConversationModel, err error) {
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupIDs", groupIDs, "groups", groups)
|
||||
}()
|
||||
return groups, utils.Wrap(getDBConn(g.DB, tx).Where("group_id in (?)", groupIDs).Find(&groups).Error, "")
|
||||
}
|
||||
|
||||
func (c *ConversationGorm) Take(ctx context.Context, groupID string) (group *table.ConversationModel, err error) {
|
||||
group = &Group{}
|
||||
defer func() {
|
||||
tracelog.SetCtxDebug(ctx, utils.GetFuncName(1), err, "groupID", groupID, "group", *group)
|
||||
}()
|
||||
return group, utils.Wrap(getDBConn(g.DB, tx).Where("group_id = ?", groupID).Take(group).Error, "")
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user