diff --git a/go.mod b/go.mod index 022546d7d..d4baf69f7 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module Open_IM -go 1.18 +go 1.17 require ( firebase.google.com/go v3.13.0+incompatible @@ -17,16 +17,11 @@ require ( github.com/aws/aws-sdk-go-v2/credentials v1.12.9 github.com/aws/aws-sdk-go-v2/service/sts v1.16.9 github.com/bwmarrin/snowflake v0.3.0 - github.com/coreos/go-systemd/v22 v22.5.0 // indirect github.com/dtm-labs/rockscache v0.0.11 github.com/fatih/structs v1.1.0 github.com/gin-gonic/gin v1.8.2 - github.com/go-openapi/spec v0.20.6 // indirect - github.com/go-openapi/swag v0.21.1 // indirect - github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/validator/v10 v10.11.1 github.com/go-redis/redis/v8 v8.11.5 - github.com/goccy/go-json v0.10.0 // indirect github.com/gogo/protobuf v1.3.2 github.com/golang-jwt/jwt/v4 v4.4.2 github.com/golang/protobuf v1.5.2 @@ -34,7 +29,6 @@ require ( github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/jinzhu/copier v0.3.5 github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible - github.com/mattn/go-isatty v0.0.17 // indirect github.com/minio/minio-go/v7 v7.0.22 github.com/mitchellh/mapstructure v1.4.2 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 @@ -51,26 +45,124 @@ require ( github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common v1.0.428 github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/sms v1.0.428 github.com/tencentyun/qcloud-cos-sts-sdk v0.0.0-20210325043845-84a0811633ca - github.com/ugorji/go/codec v1.2.8 // indirect go.etcd.io/etcd/client/v3 v3.5.6 go.mongodb.org/mongo-driver v1.8.3 - go.uber.org/atomic v1.10.0 // indirect - go.uber.org/multierr v1.9.0 // indirect - go.uber.org/zap v1.24.0 // indirect - golang.org/x/crypto v0.5.0 // indirect golang.org/x/image v0.3.0 golang.org/x/net v0.5.0 + golang.org/x/tools v0.1.12 google.golang.org/api v0.103.0 - google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect google.golang.org/grpc v1.52.0 google.golang.org/protobuf v1.28.1 - gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df - gopkg.in/ini.v1 v1.66.2 // indirect gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 gopkg.in/yaml.v3 v3.0.1 gorm.io/driver/mysql v1.3.5 gorm.io/gorm v1.23.8 ) +require ( + cloud.google.com/go v0.105.0 // indirect + cloud.google.com/go/compute v1.13.0 // indirect + cloud.google.com/go/compute/metadata v0.2.1 // indirect + cloud.google.com/go/firestore v1.9.0 // indirect + cloud.google.com/go/iam v0.8.0 // indirect + cloud.google.com/go/longrunning v0.3.0 // indirect + cloud.google.com/go/storage v1.27.0 // indirect + github.com/KyleBanks/depth v1.2.1 // indirect + github.com/OpenIMSDK/open_log v1.0.0 // indirect + github.com/alibabacloud-go/debug v0.0.0-20190504072949-9472017b5c68 // indirect + github.com/alibabacloud-go/endpoint-util v1.1.0 // indirect + github.com/alibabacloud-go/openapi-util v0.0.9 // indirect + github.com/alibabacloud-go/tea-utils v1.3.9 // indirect + github.com/aliyun/credentials-go v1.1.2 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.8 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.14 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.8 // indirect + github.com/aws/aws-sdk-go-v2/internal/ini v1.3.15 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.8 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.11.12 // indirect + github.com/aws/smithy-go v1.12.0 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.1.2 // indirect + github.com/coreos/go-semver v0.3.0 // indirect + github.com/coreos/go-systemd/v22 v22.5.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/dustin/go-humanize v1.0.0 // indirect + github.com/eapache/go-resiliency v1.2.0 // indirect + github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect + github.com/eapache/queue v1.1.0 // indirect + github.com/gin-contrib/sse v0.1.0 // indirect + github.com/go-openapi/jsonpointer v0.19.5 // indirect + github.com/go-openapi/jsonreference v0.20.0 // indirect + github.com/go-openapi/spec v0.20.6 // indirect + github.com/go-openapi/swag v0.21.1 // indirect + github.com/go-playground/locales v0.14.1 // indirect + github.com/go-playground/universal-translator v0.18.0 // indirect + github.com/go-sql-driver/mysql v1.6.0 // indirect + github.com/go-stack/stack v1.8.0 // indirect + github.com/goccy/go-json v0.10.0 // indirect + github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect + github.com/golang/snappy v0.0.3 // indirect + github.com/google/go-cmp v0.5.9 // indirect + github.com/google/uuid v1.3.0 // indirect + github.com/googleapis/enterprise-certificate-proxy v0.2.0 // indirect + github.com/googleapis/gax-go/v2 v2.7.0 // indirect + github.com/hashicorp/go-uuid v1.0.2 // indirect + github.com/jcmturner/aescts/v2 v2.0.0 // indirect + github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect + github.com/jcmturner/gofork v1.0.0 // indirect + github.com/jcmturner/gokrb5/v8 v8.4.2 // indirect + github.com/jcmturner/rpc/v2 v2.0.3 // indirect + github.com/jinzhu/inflection v1.0.0 // indirect + github.com/jinzhu/now v1.1.5 // indirect + github.com/josharian/intern v1.0.0 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/compress v1.13.6 // indirect + github.com/klauspost/cpuid v1.3.1 // indirect + github.com/leodido/go-urn v1.2.1 // indirect + github.com/lestrrat-go/strftime v1.0.6 // indirect + github.com/lithammer/shortuuid v3.0.0+incompatible // indirect + github.com/mailru/easyjson v0.7.7 // indirect + github.com/mattn/go-isatty v0.0.17 // indirect + github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect + github.com/minio/md5-simd v1.1.0 // indirect + github.com/minio/sha256-simd v0.1.1 // indirect + github.com/mitchellh/go-homedir v1.1.0 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pelletier/go-toml/v2 v2.0.6 // indirect + github.com/pierrec/lz4 v2.6.0+incompatible // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/prometheus/client_model v0.2.0 // indirect + github.com/prometheus/common v0.37.0 // indirect + github.com/prometheus/procfs v0.8.0 // indirect + github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect + github.com/rs/xid v1.2.1 // indirect + github.com/tjfoc/gmsm v1.3.2 // indirect + github.com/ugorji/go/codec v1.2.8 // indirect + github.com/xdg-go/pbkdf2 v1.0.0 // indirect + github.com/xdg-go/scram v1.0.2 // indirect + github.com/xdg-go/stringprep v1.0.2 // indirect + github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect + go.etcd.io/etcd/api/v3 v3.5.6 // indirect + go.etcd.io/etcd/client/pkg/v3 v3.5.6 // indirect + go.opencensus.io v0.24.0 // indirect + go.uber.org/atomic v1.10.0 // indirect + go.uber.org/multierr v1.9.0 // indirect + go.uber.org/zap v1.24.0 // indirect + golang.org/x/crypto v0.5.0 // indirect + golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783 // indirect + golang.org/x/sync v0.1.0 // indirect + golang.org/x/sys v0.4.0 // indirect + golang.org/x/text v0.6.0 // indirect + golang.org/x/time v0.1.0 // indirect + golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect + google.golang.org/appengine v1.6.7 // indirect + google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect + gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect + gopkg.in/ini.v1 v1.66.2 // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect +) + replace github.com/Shopify/sarama => github.com/Shopify/sarama v1.29.0 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/constant/constant.go b/pkg/common/constant/constant.go index f2d62c0d9..7bff14efa 100644 --- a/pkg/common/constant/constant.go +++ b/pkg/common/constant/constant.go @@ -303,7 +303,7 @@ const ( BecomeFriendByImport = 1 //管理员导入 BecomeFriendByMyApply = 2 //自己主动申请添加 BecomeFriendByPeerApply = 3 //对方主动申请添加 - BecomeFriendByApply = 4 //自己主动申请添加 + BecomeFriendByApply = 4 //申请添加 ) const ( 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/controller/friend.go b/pkg/common/db/controller/friend.go index bf143a3fa..04bc180dd 100644 --- a/pkg/common/db/controller/friend.go +++ b/pkg/common/db/controller/friend.go @@ -2,8 +2,9 @@ package controller import ( "Open_IM/pkg/common/constant" - "Open_IM/pkg/common/db/relation" - relation2 "Open_IM/pkg/common/db/table/relation" + relation1 "Open_IM/pkg/common/db/relation" + "Open_IM/pkg/common/db/table/relation" + "Open_IM/pkg/utils" "context" "gorm.io/gorm" ) @@ -14,25 +15,25 @@ type FriendInterface interface { // AddFriendRequest 增加或者更新好友申请 AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error) // BecomeFriend 先判断是否在好友表，如果在则不插入 - BecomeFriend(ctx context.Context, friends []*relation2.FriendModel, revFriends []*relation2.FriendModel) (err error) + BecomeFriend(ctx context.Context, friends []*relation.FriendModel, revFriends []*relation.FriendModel) (err error) // RefuseFriendRequest 拒绝好友申请 - RefuseFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) + RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) // AgreeFriendRequest 同意好友申请 - AgreeFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) + AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) // Delete 删除好友 Delete(ctx context.Context, ownerUserID string, friendUserIDs string) (err error) // UpdateRemark 更新好友备注 UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) // FindOwnerFriends 获取ownerUserID的好友列表 - FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, total int64, err error) + FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) // FindInWhoseFriends friendUserID在哪些人的好友列表中 - FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, total int64, err error) + FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) // FindFriendRequestFromMe 获取我发出去的好友申请 - FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, total int64, err error) + FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) // FindFriendRequestToMe 获取我收到的的好友申请 - FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, total int64, err error) + FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) // FindFriends 获取某人指定好友的信息 如果有一个不存在也返回错误 - FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation2.FriendModel, err error) + FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error) } type FriendController struct { @@ -44,51 +45,64 @@ func NewFriendController(db *gorm.DB) *FriendController { } // CheckIn 检查user2是否在user1的好友列表中(inUser1Friends==true) 检查user1是否在user2的好友列表中(inUser2Friends==true) -func (f *FriendController) CheckIn(ctx context.Context, user1, user2 string) (err error, inUser1Friends bool, inUser2Friends bool) { +func (f *FriendController) CheckIn(ctx context.Context, user1, user2 string) (inUser1Friends bool, inUser2Friends bool, err error) { + return f.database.CheckIn(ctx, user1, user2) } // AddFriendRequest 增加或者更新好友申请 func (f *FriendController) AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error) { + return f.database.AddFriendRequest(ctx, fromUserID, toUserID, reqMsg, ex) } // BecomeFriend 先判断是否在好友表，如果在则不插入 -func (f *FriendController) BecomeFriend(ctx context.Context, friends []*relation2.FriendModel) (err error) { +func (f *FriendController) BecomeFriend(ctx context.Context, ownerUserID string, friends []*relation.FriendModel) (err error) { + return f.database.BecomeFriend(ctx, ownerUserID, friends) } // RefuseFriendRequest 拒绝好友申请 -func (f *FriendController) RefuseFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) { +func (f *FriendController) RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) { + return f.database.RefuseFriendRequest(ctx, friendRequest) } // AgreeFriendRequest 同意好友申请 -func (f *FriendController) AgreeFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) { +func (f *FriendController) AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) { + return f.database.AgreeFriendRequest(ctx, friendRequest) } // Delete 删除好友 func (f *FriendController) Delete(ctx context.Context, ownerUserID string, friendUserIDs string) (err error) { + return f.database.Delete(ctx, ownerUserID, friendUserIDs) } // UpdateRemark 更新好友备注 func (f *FriendController) UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) { + return f.database.UpdateRemark(ctx, ownerUserID, friendUserID, remark) } // FindOwnerFriends 获取ownerUserID的好友列表 -func (f *FriendController) FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, err error) { +func (f *FriendController) FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) { + return f.database.FindOwnerFriends(ctx, ownerUserID, pageNumber, showNumber) } // FindInWhoseFriends friendUserID在哪些人的好友列表中 -func (f *FriendController) FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, err error) { +func (f *FriendController) FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) { + return f.database.FindInWhoseFriends(ctx, friendUserID, pageNumber, showNumber) } // FindFriendRequestFromMe 获取我发出去的好友申请 -func (f *FriendController) FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, err error) { +func (f *FriendController) FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) { + return f.database.FindFriendRequestFromMe(ctx, userID, pageNumber, showNumber) } // FindFriendRequestToMe 获取我收到的的好友申请 -func (f *FriendController) FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, err error) { +func (f *FriendController) FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) { + return f.database.FindFriendRequestToMe(ctx, userID, pageNumber, showNumber) } // FindFriends 获取某人指定好友的信息 -func (f *FriendController) FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation2.FriendModel, err error) { +func (f *FriendController) FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error) { + return f.database.FindFriends(ctx, ownerUserID, friendUserIDs) + } type FriendDatabaseInterface interface { @@ -97,34 +111,34 @@ type FriendDatabaseInterface interface { // AddFriendRequest 增加或者更新好友申请 AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error) // BecomeFriend 先判断是否在好友表，如果在则不插入 - BecomeFriend(ctx context.Context, friends []*relation2.FriendModel) (err error) + BecomeFriend(ctx context.Context, ownerUserID string, friends []*relation.FriendModel) (err error) // RefuseFriendRequest 拒绝好友申请 - RefuseFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) + RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) // AgreeFriendRequest 同意好友申请 - AgreeFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) + AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) // Delete 删除好友 Delete(ctx context.Context, ownerUserID string, friendUserIDs string) (err error) // UpdateRemark 更新好友备注 UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) // FindOwnerFriends 获取ownerUserID的好友列表 - FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, err error) + FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) // FindInWhoseFriends friendUserID在哪些人的好友列表中 - FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, err error) + FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) // FindFriendRequestFromMe 获取我发出去的好友申请 - FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, err error) + FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) // FindFriendRequestToMe 获取我收到的的好友申请 - FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, err error) + FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) // FindFriends 获取某人指定好友的信息 - FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation2.FriendModel, err error) + FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error) } type FriendDatabase struct { - friend *relation.FriendGorm - friendRequest *relation.FriendRequestGorm + friend *relation1.FriendGorm + friendRequest *relation1.FriendRequestGorm } func NewFriendDatabase(db *gorm.DB) *FriendDatabase { - return &FriendDatabase{friend: relation.NewFriendGorm(db), friendRequest: relation.NewFriendRequestGorm(db)} + return &FriendDatabase{friend: relation1.NewFriendGorm(db), friendRequest: relation1.NewFriendRequestGorm(db)} } // CheckIn 检查user2是否在user1的好友列表中(inUser1Friends==true) 检查user1是否在user2的好友列表中(inUser2Friends==true) @@ -143,11 +157,31 @@ func (f *FriendDatabase) CheckIn(ctx context.Context, userID1, userID2 string) ( // AddFriendRequest 增加或者更新好友申请 func (f *FriendDatabase) AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error) { - + return f.friendRequest.DB.Transaction(func(tx *gorm.DB) error { + fq, err := f.friendRequest.Find(ctx, fromUserID, toUserID, tx) + if err != nil { + return err + } + if fq != nil { // + m := make(map[string]interface{}, 1) + m["handle_result"] = 0 + m["handle_msg"] = "" + m["req_msg"] = reqMsg + m["ex"] = ex + if err := f.friendRequest.UpdateByMap(ctx, fromUserID, toUserID, m, tx); err != nil { + return err + } + } else { + if err := f.friendRequest.Create(ctx, []*relation.FriendRequestModel{&relation.FriendRequestModel{FromUserID: fromUserID, ToUserID: toUserID, ReqMsg: reqMsg, Ex: ex}}, tx); err != nil { + return err + } + } + return nil + }) } // BecomeFriend 先判断是否在好友表，如果在则不插入 -func (f *FriendDatabase) BecomeFriend(ctx context.Context, ownerUserID string, friends []*relation2.FriendModel) (err error) { +func (f *FriendDatabase) BecomeFriend(ctx context.Context, ownerUserID string, friends []*relation.FriendModel) (err error) { return f.friend.DB.Transaction(func(tx *gorm.DB) error { //先find 找出重复的 去掉重复的 friendUserIDs := make([]string, 0, len(friends)) @@ -158,51 +192,117 @@ func (f *FriendDatabase) BecomeFriend(ctx context.Context, ownerUserID string, f if err != nil { return err } + fs1 = append(fs1, friends...) + fs11 := utils.DistinctAny(fs1, func(e *relation.FriendModel) string { + return utils.UniqueJoin(e.OwnerUserID, e.FriendUserID) + }) + err = f.friend.Create(ctx, fs11, tx) + if err != nil { + return err + } fs2, err := f.friend.FindReversalFriends(ctx, ownerUserID, friendUserIDs, tx) if err != nil { return err } - + for _, v := range friends { + fs2 = append(fs2, &relation.FriendModel{OwnerUserID: v.FriendUserID, FriendUserID: ownerUserID}) + } + fs22 := utils.DistinctAny(fs2, func(e *relation.FriendModel) string { + return utils.UniqueJoin(e.OwnerUserID, e.FriendUserID) + }) + err = f.friend.Create(ctx, fs22, tx) + if err != nil { + return err + } return nil }) - } // RefuseFriendRequest 拒绝好友申请 -func (f *FriendDatabase) RefuseFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) { +func (f *FriendDatabase) RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) { + friendRequest.HandleResult = constant.FriendResponseRefuse + err = f.friendRequest.Update(ctx, []*relation.FriendRequestModel{friendRequest}) + if err != nil { + return err + } + return nil } // AgreeFriendRequest 同意好友申请 -func (f *FriendDatabase) AgreeFriendRequest(ctx context.Context, friendRequest *relation2.FriendRequestModel) (err error) { +func (f *FriendDatabase) AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) { + return f.friend.DB.Transaction(func(tx *gorm.DB) error { + //先find 找出重复的 去掉重复的 + fs1, err := f.friend.FindFriends(ctx, friendRequest.FromUserID, []string{friendRequest.ToUserID}, tx) + if err != nil { + return err + } + if len(fs1) == 0 { + err = f.friend.Create(ctx, []*relation.FriendModel{ + &relation.FriendModel{ + OwnerUserID: friendRequest.FromUserID, + FriendUserID: friendRequest.ToUserID, + OperatorUserID: friendRequest.ToUserID, + AddSource: constant.BecomeFriendByApply, + }}, tx) + if err != nil { + return err + } + } + fs2, err := f.friend.FindReversalFriends(ctx, friendRequest.ToUserID, []string{friendRequest.FromUserID}, tx) + if len(fs2) == 0 { + err = f.friend.Create(ctx, []*relation.FriendModel{ + &relation.FriendModel{ + OwnerUserID: friendRequest.ToUserID, + FriendUserID: friendRequest.FromUserID, + OperatorUserID: friendRequest.ToUserID, + AddSource: constant.BecomeFriendByApply, + }}, tx) + if err != nil { + return err + } + } + friendRequest.HandleResult = constant.FriendResponseAgree + err = f.friendRequest.Update(ctx, []*relation.FriendRequestModel{friendRequest}, tx) + if err != nil { + return err + } + return nil + }) } // Delete 删除好友 -func (f *FriendDatabase) Delete(ctx context.Context, ownerUserID string, friendUserIDs string) (err error) { +func (f *FriendDatabase) Delete(ctx context.Context, ownerUserID string, friendUserID string) (err error) { + return f.friend.Delete(ctx, ownerUserID, friendUserID) } // UpdateRemark 更新好友备注 func (f *FriendDatabase) UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) { + return f.friend.UpdateRemark(ctx, ownerUserID, friendUserID, remark) } // FindOwnerFriends 获取ownerUserID的好友列表 -func (f *FriendDatabase) FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, err error) { +func (f *FriendDatabase) FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) { + return f.friend.FindOwnerFriends(ctx, ownerUserID, pageNumber, showNumber) } // FindInWhoseFriends friendUserID在哪些人的好友列表中 -func (f *FriendDatabase) FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation2.FriendModel, err error) { +func (f *FriendDatabase) FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) { + return f.friend.FindInWhoseFriends(ctx, friendUserID, pageNumber, showNumber) } // FindFriendRequestFromMe 获取我发出去的好友申请 -func (f *FriendDatabase) FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, err error) { +func (f *FriendDatabase) FindFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) { + return f.friendRequest.FindFromUserID(ctx, userID, pageNumber, showNumber) } // FindFriendRequestToMe 获取我收到的的好友申请 -func (f *FriendDatabase) FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation2.FriendRequestModel, err error) { +func (f *FriendDatabase) FindFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) { + return f.friendRequest.FindToUserID(ctx, userID, pageNumber, showNumber) } -// FindFriends 获取某人指定好友的信息 如果有一个不存在也返回错误 -func (f *FriendDatabase) FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation2.FriendModel, err error) { - friends, err = f.friend.Find(ctx, ownerUserID, friendUserIDs) +// FindFriends 获取某人指定好友的信息 如果有一个不存在，也返回错误 +func (f *FriendDatabase) FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error) { + friends, err = f.friend.FindFriends(ctx, ownerUserID, friendUserIDs) if err != nil { return } 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, "") +} diff --git a/pkg/common/db/relation/friend_model_k.go b/pkg/common/db/relation/friend_model_k.go index fa5bb5fdb..03c642cc4 100644 --- a/pkg/common/db/relation/friend_model_k.go +++ b/pkg/common/db/relation/friend_model_k.go @@ -34,14 +34,14 @@ func (f *FriendGorm) Create(ctx context.Context, friends []*relation.FriendModel defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friends", friends) }() - return utils.Wrap(getDBConn(f.DB, tx).Model(&relation.FriendModel{}).Create(&friends).Error, "") + return utils.Wrap(getDBConn(f.DB, tx).Model(&table.FriendModel{}).Create(&friends).Error, "") } func (f *FriendGorm) Delete(ctx context.Context, ownerUserID string, friendUserIDs string) (err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "friendUserIDs", friendUserIDs) }() - err = utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("owner_user_id = ? and friend_user_id = ?", ownerUserID, friendUserIDs).Delete(&relation.FriendModel{}).Error, "") + err = utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("owner_user_id = ? and friend_user_id = ?", ownerUserID, friendUserIDs).Delete(&table.FriendModel{}).Error, "") return err } @@ -49,51 +49,51 @@ func (f *FriendGorm) UpdateByMap(ctx context.Context, ownerUserID string, args m defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "args", args) }() - return utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("owner_user_id = ?", ownerUserID).Updates(args).Error, "") + return utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("owner_user_id = ?", ownerUserID).Updates(args).Error, "") } func (f *FriendGorm) Update(ctx context.Context, friends []*relation.FriendModel, tx ...*gorm.DB) (err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friends", friends) }() - return utils.Wrap(f.DB.Model(&relation.FriendModel{}).Updates(&friends).Error, "") + return utils.Wrap(f.DB.Model(&table.FriendModel{}).Updates(&friends).Error, "") } func (f *FriendGorm) UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "friendUserID", friendUserID, "remark", remark) }() - return utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("owner_user_id = ? and friend_user_id = ?", ownerUserID, friendUserID).Update("remark", remark).Error, "") + return utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("owner_user_id = ? and friend_user_id = ?", ownerUserID, friendUserID).Update("remark", remark).Error, "") } func (f *FriendGorm) FindOwnerUserID(ctx context.Context, ownerUserID string) (friends []*relation.FriendModel, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "friends", friends) }() - return friends, utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("owner_user_id = ?", ownerUserID).Find(&friends).Error, "") + return friends, utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("owner_user_id = ?", ownerUserID).Find(&friends).Error, "") } func (f *FriendGorm) FindFriendUserID(ctx context.Context, friendUserID string) (friends []*relation.FriendModel, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friendUserID", friendUserID, "friends", friends) }() - return friends, utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("friend_user_id = ?", friendUserID).Find(&friends).Error, "") + return friends, utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("friend_user_id = ?", friendUserID).Find(&friends).Error, "") } func (f *FriendGorm) Take(ctx context.Context, ownerUserID, friendUserID string) (friend *relation.FriendModel, err error) { - friend = &relation.FriendModel{} + friend = &table.FriendModel{} defer tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "friendUserID", friendUserID, "friend", friend) - return friend, utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("owner_user_id = ? and friend_user_id", ownerUserID, friendUserID).Take(friend).Error, "") + return friend, utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("owner_user_id = ? and friend_user_id", ownerUserID, friendUserID).Take(friend).Error, "") } func (f *FriendGorm) FindUserState(ctx context.Context, userID1, userID2 string) (friends []*relation.FriendModel, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "userID1", userID1, "userID2", userID2) }() - return friends, utils.Wrap(f.DB.Model(&relation.FriendModel{}).Where("(owner_user_id = ? and friend_user_id = ?) or (owner_user_id = ? and friend_user_id = ?)", userID1, userID2, userID2, userID1).Find(&friends).Error, "") + return friends, utils.Wrap(f.DB.Model(&table.FriendModel{}).Where("(owner_user_id = ? and friend_user_id = ?) or (owner_user_id = ? and friend_user_id = ?)", userID1, userID2, userID2, userID1).Find(&friends).Error, "") } -// 获取 owner的好友列表 +// 获取 owner的好友列表 如果不存在也不返回错误 func (f *FriendGorm) FindFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, tx ...*gorm.DB) (friends []*relation.FriendModel, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friendUserIDs", friendUserIDs, "friends", friends) @@ -101,10 +101,34 @@ func (f *FriendGorm) FindFriends(ctx context.Context, ownerUserID string, friend return friends, utils.Wrap(getDBConn(f.DB, tx).Where("owner_user_id = ? AND friend_user_id in (?)", ownerUserID, friendUserIDs).Find(&friends).Error, "") } -// 获取哪些人添加了friendUserID +// 获取哪些人添加了friendUserID 如果不存在也不返回错误 func (f *FriendGorm) FindReversalFriends(ctx context.Context, friendUserID string, ownerUserIDs []string, tx ...*gorm.DB) (friends []*relation.FriendModel, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friendUserID", friendUserID, "friends", friends) }() return friends, utils.Wrap(getDBConn(f.DB, tx).Where("friend_user_id = ? AND owner_user_id in (?)", friendUserID, ownerUserIDs).Find(&friends).Error, "") } + +func (f *FriendGorm) FindOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32, tx ...*gorm.DB) (friends []*relation.FriendModel, total int64, err error) { + defer func() { + tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "pageNumber", pageNumber, "showNumber", showNumber, "friends", friends, "total", total) + }() + err = getDBConn(f.DB, tx).Model(f).Where("owner_user_id = ? ", ownerUserID).Count(&total).Error + if err != nil { + return nil, 0, utils.Wrap(err, "") + } + err = utils.Wrap(getDBConn(f.DB, tx).Model(f).Where("owner_user_id = ? ", ownerUserID).Limit(int(showNumber)).Offset(int(pageNumber*showNumber)).Find(&friends).Error, "") + return +} + +func (f *FriendGorm) FindInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32, tx ...*gorm.DB) (friends []*relation.FriendModel, total int64, err error) { + defer func() { + tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friendUserID", friendUserID, "pageNumber", pageNumber, "showNumber", showNumber, "friends", friends, "total", total) + }() + err = getDBConn(f.DB, tx).Model(f).Where("friend_user_id = ? ", friendUserID).Count(&total).Error + if err != nil { + return nil, 0, utils.Wrap(err, "") + } + err = utils.Wrap(getDBConn(f.DB, tx).Model(f).Where("friend_user_id = ? ", friendUserID).Limit(int(showNumber)).Offset(int(pageNumber*showNumber)).Find(&friends).Error, "") + return +} diff --git a/pkg/common/db/relation/friend_request_model.go b/pkg/common/db/relation/friend_request_model.go index f5a1261f4..9a2fa3365 100644 --- a/pkg/common/db/relation/friend_request_model.go +++ b/pkg/common/db/relation/friend_request_model.go @@ -20,7 +20,7 @@ type FriendRequestGorm struct { DB *gorm.DB `gorm:"-"` } -func (f *FriendRequestGorm) Create(ctx context.Context, friends []*relation.FriendRequestModel) (err error) { +func (f *FriendRequestGorm) Create(ctx context.Context, friends []*relation.FriendRequestModel, tx ...*gorm.DB) (err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friends", friends) }() @@ -34,25 +34,18 @@ func (f *FriendRequestGorm) Delete(ctx context.Context, fromUserID, toUserID str return utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Where("from_user_id = ? and to_user_id = ?", fromUserID, toUserID).Delete(&relation.FriendRequestModel{}).Error, "") } -func (f *FriendRequestGorm) UpdateByMap(ctx context.Context, ownerUserID string, args map[string]interface{}) (err error) { +func (f *FriendRequestGorm) UpdateByMap(ctx context.Context, formUserID string, toUserID string, args map[string]interface{}, tx ...*gorm.DB) (err error) { defer func() { - tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "args", args) + tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "formUserID", formUserID, "toUserID", toUserID, "args", args) }() - return utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Where("owner_user_id = ?", ownerUserID).Updates(args).Error, "") + return utils.Wrap(getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Where("from_user_id = ? AND to_user_id ", formUserID, toUserID).Updates(args).Error, "") } -func (f *FriendRequestGorm) Update(ctx context.Context, friends []*relation.FriendRequestModel) (err error) { +func (f *FriendRequestGorm) Update(ctx context.Context, friendRequests []*relation.FriendRequestModel, tx ...*gorm.DB) (err error) { defer func() { - tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friends", friends) + tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "friendRequests", friendRequests) }() - return utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Updates(&friends).Error, "") -} - -func (f *FriendRequestGorm) Find(ctx context.Context, ownerUserID string) (friends []*relation.FriendRequestModel, err error) { - defer func() { - tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "ownerUserID", ownerUserID, "friends", friends) - }() - return friends, utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Where("owner_user_id = ?", ownerUserID).Find(&friends).Error, "") + return utils.Wrap(getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Updates(&friendRequests).Error, "") } func (f *FriendRequestGorm) Take(ctx context.Context, fromUserID, toUserID string) (friend *relation.FriendRequestModel, err error) { @@ -61,16 +54,34 @@ func (f *FriendRequestGorm) Take(ctx context.Context, fromUserID, toUserID strin return friend, utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Where("from_user_id = ? and to_user_id", fromUserID, toUserID).Take(friend).Error, "") } -func (f *FriendRequestGorm) FindToUserID(ctx context.Context, toUserID string) (friends []*relation.FriendRequestModel, err error) { +func (f *FriendRequestGorm) Find(ctx context.Context, fromUserID, toUserID string, tx ...*gorm.DB) (friend *relation.FriendRequestModel, err error) { + friend = &relation.FriendRequestModel{} + defer tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "fromUserID", fromUserID, "toUserID", toUserID, "friend", friend) + return friend, utils.Wrap(getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Where("from_user_id = ? and to_user_id", fromUserID, toUserID).Find(friend).Error, "") +} + +func (f *FriendRequestGorm) FindToUserID(ctx context.Context, toUserID string, pageNumber, showNumber int32, tx ...*gorm.DB) (friends []*relation.FriendRequestModel, total int64, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "toUserID", toUserID, "friends", friends) }() - return friends, utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Where("to_user_id = ?", toUserID).Find(&friends).Error, "") + + err = getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Where("to_user_id = ? ", toUserID).Count(&total).Error + if err != nil { + return nil, 0, utils.Wrap(err, "") + } + err = utils.Wrap(getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Where("to_user_id = ? ", toUserID).Limit(int(showNumber)).Offset(int(pageNumber*showNumber)).Find(&friends).Error, "") + return } -func (f *FriendRequestGorm) FindFromUserID(ctx context.Context, fromUserID string) (friends []*relation.FriendRequestModel, err error) { +func (f *FriendRequestGorm) FindFromUserID(ctx context.Context, fromUserID string, pageNumber, showNumber int32, tx ...*gorm.DB) (friends []*relation.FriendRequestModel, total int64, err error) { defer func() { tracelog.SetCtxDebug(ctx, utils.GetSelfFuncName(), err, "fromUserID", fromUserID, "friends", friends) }() - return friends, utils.Wrap(f.DB.Model(&relation.FriendRequestModel{}).Where("from_user_id = ?", fromUserID).Find(&friends).Error, "") + + err = getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Where("from_user_id = ? ", fromUserID).Count(&total).Error + if err != nil { + return nil, 0, utils.Wrap(err, "") + } + err = utils.Wrap(getDBConn(f.DB, tx).Model(&relation.FriendRequestModel{}).Where("from_user_id = ? ", fromUserID).Limit(int(showNumber)).Offset(int(pageNumber*showNumber)).Find(&friends).Error, "") + return }