diff --git a/internal/common/check/group.go b/internal/common/check/group.go new file mode 100644 index 000000000..56df95e76 --- /dev/null +++ b/internal/common/check/group.go @@ -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") +} diff --git a/internal/rpc/conversation/conversaion.go b/internal/rpc/conversation/conversaion.go index e9b16387b..bacba5202 100644 --- a/internal/rpc/conversation/conversaion.go +++ b/internal/rpc/conversation/conversaion.go @@ -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 +} diff --git a/internal/rpc/msg/conversation_notification.go b/internal/rpc/msg/conversation_notification.go index 22e1dedc7..a62d48d4e 100644 --- a/internal/rpc/msg/conversation_notification.go +++ b/internal/rpc/msg/conversation_notification.go @@ -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, diff --git a/pkg/common/db/cache/conversation.go b/pkg/common/db/cache/conversation.go index 0a4dd3711..d7757397e 100644 --- a/pkg/common/db/cache/conversation.go +++ b/pkg/common/db/cache/conversation.go @@ -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") } diff --git a/pkg/common/db/controller/conversation.go b/pkg/common/db/controller/conversation.go new file mode 100644 index 000000000..3ed221c45 --- /dev/null +++ b/pkg/common/db/controller/conversation.go @@ -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 +//} diff --git a/pkg/common/db/relation/conversation_model.go b/pkg/common/db/relation/conversation_model.go index d8c730607..1071e26af 100644 --- a/pkg/common/db/relation/conversation_model.go +++ b/pkg/common/db/relation/conversation_model.go @@ -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 diff --git a/pkg/common/db/relation/conversation_model_g.go b/pkg/common/db/relation/conversation_model_g.go new file mode 100644 index 000000000..37882940d --- /dev/null +++ b/pkg/common/db/relation/conversation_model_g.go @@ -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, "") +}