blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-01-11 06:56:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

update get chat logs api

Browse Source 
Signed-off-by: ‘hanzhixiao’ <‘709674996@qq.com’>
This commit is contained in:
‘hanzhixiao’ 2023-07-11 18:26:31 +08:00
parent a87ae64acd
commit 929e969af1
17 changed files with 865 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/api/main.go
View File

@ -42,7 +42,6 @@ func startPprof() {
}
func run(port int) error {
port = 10002
if port == 0 {
return fmt.Errorf("port is empty")
}

6
config/config.yaml
View File

@ -11,7 +11,7 @@ zookeeper:
mysql:
address: [ 127.0.0.1:3306 ] #目前仅支持单机
username: root #用户名
password: #密码
password: openIM123 #密码
database: openIM_v3 #不建议修改
maxOpenConn: 1000 #最大连接数
maxIdleConn: 100 #最大空闲连接数
@ -28,7 +28,7 @@ mongo:
maxPoolSize: 100
redis:
address: [ 127.0.0.1:6379 ] #
address: [ 127.0.0.1:16379 ] #
username: #only redis version 6.0+ need username
password: openIM123 #密码
@ -68,7 +68,7 @@ object:
tempBucket: "openim" #不建议修改
dataBucket: "openim" #不建议修改
location: us-east-1 #不建议修改
endpoint: http://192.168.245.231:10005 #minio对外服务的ip和端口，app要能访问此ip和端口
endpoint: http://127.0.0.1:10005 #minio对外服务的ip和端口，app要能访问此ip和端口
accessKeyID: root #ID
secretAccessKey: openIM123 #秘钥
isDistributedMod: false #是否分布式多硬盘部署，如果是多硬盘部署，需要修改为true

4
internal/api/msg.go
View File

@ -350,3 +350,7 @@ func (m *MessageApi) GetActiveGroup(c *gin.Context) {
func (m *MessageApi) SearchMsg(c *gin.Context) {
a2r.Call(msg.MsgClient.SearchMessage, m.Client, c)
}
func (m MessageApi) GetChatLogs(c *gin.Context) {
a2r.Call(msg.MsgClient.GetChatLogs, m.Client, c)
}

1
internal/api/route.go
View File

@ -140,6 +140,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
msgGroup.POST("/batch_send_msg", m.ManagementBatchSendMsg)
msgGroup.POST("/check_msg_is_send_success", m.CheckMsgIsSendSuccess)
msgGroup.POST("/search_msg", m.SearchMsg)
msgGroup.POST("/get_chat_logs", m.GetChatLogs)
}
//Conversation
conversationGroup := r.Group("/conversation", ParseToken)

2
internal/msggateway/init.go
View File

@ -8,8 +8,6 @@ import (
)
func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
rpcPort = 10040
wsPort = 10001
fmt.Println("start rpc/msg_gateway server, port: ", rpcPort, wsPort, prometheusPort, ", OpenIM version: ", config.Version)
longServer, err := NewWsServer(
WithPort(wsPort),

5
internal/msgtransfer/init.go
View File

@ -59,11 +59,12 @@ func StartTransfer(prometheusPort int) error {
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
msgModel := cache.NewMsgCacheModel(rdb)
msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
msgMysModel := relation.NewChatLogGorm(db)
extendMsgModel := unrelation.NewExtendMsgSetMongoDriver(mongo.GetDatabase())
extendMsgCache := cache.NewExtendMsgSetCacheRedis(rdb, extendMsgModel, cache.GetDefaultOpt())
chatLogDatabase := controller.NewChatLogDatabase(relation.NewChatLogGorm(db))
chatLogDatabase := controller.NewChatLogDatabase(msgMysModel)
extendMsgDatabase := controller.NewExtendMsgDatabase(extendMsgModel, extendMsgCache, tx.NewMongo(mongo.GetClient()))
msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)
msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel, msgMysModel)
conversationRpcClient := rpcclient.NewConversationRpcClient(client)
groupRpcClient := rpcclient.NewGroupRpcClient(client)
msgTransfer := NewMsgTransfer(chatLogDatabase, extendMsgDatabase, msgDatabase, &conversationRpcClient, &groupRpcClient)

5
internal/rpc/msg/server.go
View File

@ -2,6 +2,7 @@ package msg
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/relation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
@ -57,15 +58,17 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
if err != nil {
return err
}
mysql, err := relation.NewGormDB()
if err := mongo.CreateMsgIndex(); err != nil {
return err
}
cacheModel := cache.NewMsgCacheModel(rdb)
msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
msgMysModel := relation.NewChatLogGorm(mysql)
extendMsgModel := unrelation.NewExtendMsgSetMongoDriver(mongo.GetDatabase())
extendMsgCacheModel := cache.NewExtendMsgSetCacheRedis(rdb, extendMsgModel, cache.GetDefaultOpt())
extendMsgDatabase := controller.NewExtendMsgDatabase(extendMsgModel, extendMsgCacheModel, tx.NewMongo(mongo.GetClient()))
msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, cacheModel)
msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, cacheModel, msgMysModel)
conversationClient := rpcclient.NewConversationRpcClient(client)
userRpcClient := rpcclient.NewUserRpcClient(client)
groupRpcClient := rpcclient.NewGroupRpcClient(client)

56
internal/rpc/msg/sync_msg.go
View File

@ -2,6 +2,7 @@ package msg
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
@ -88,3 +89,58 @@ func (m *msgServer) GetMaxSeq(ctx context.Context, req *sdkws.GetMaxSeqReq) (*sd
resp.MaxSeqs = maxSeqs
return resp, nil
}
func (m *msgServer) GetChatLogs(ctx context.Context, req *msg.GetChatLogsReq) (*msg.GetChatLogsResp, error) {
resp := &msg.GetChatLogsResp{}
num, chatLogs, err := m.MsgDatabase.GetChatLog(ctx, req, req.Pagination.PageNumber, req.Pagination.ShowNumber, []int32{
constant.Text,
constant.Picture,
constant.Voice,
constant.Video,
constant.File,
constant.AtText,
constant.Merger,
constant.Card,
constant.Location,
constant.Custom,
constant.Revoke,
constant.Quote,
constant.AdvancedText,
constant.CustomNotTriggerConversation,
})
if err != nil {
return nil, err
}
resp.ChatLogsNum = int32(num)
for _, chatLog := range chatLogs {
pbChatLog := &msg.ChatLog{}
utils.CopyStructFields(pbChatLog, chatLog)
pbChatLog.SendTime = chatLog.SendTime.Unix()
pbChatLog.CreateTime = chatLog.CreateTime.Unix()
if chatLog.SenderNickname == "" {
sendUser, err := m.User.GetUserInfo(ctx, chatLog.SendID)
if err != nil {
return nil, err
}
pbChatLog.SenderNickname = sendUser.Nickname
}
switch chatLog.SessionType {
case constant.SingleChatType:
recvUser, err := m.User.GetUserInfo(ctx, chatLog.RecvID)
if err != nil {
return nil, err
}
pbChatLog.SenderNickname = recvUser.Nickname
case constant.GroupChatType, constant.SuperGroupChatType:
group, err := m.Group.GetGroupInfo(ctx, chatLog.RecvID)
if err != nil {
continue
}
pbChatLog.RecvID = group.GroupID
pbChatLog.GroupName = group.GroupName
}
resp.ChatLogs = append(resp.ChatLogs, pbChatLog)
}
return resp, nil
}

2
internal/tools/msg.go
View File

@ -49,7 +49,7 @@ func InitMsgTool() (*MsgTool, error) {
return nil, err
}
userDB := relation.NewUserGorm(db)
msgDatabase := controller.InitCommonMsgDatabase(rdb, mongo.GetDatabase())
msgDatabase := controller.InitCommonMsgDatabase(rdb, mongo.GetDatabase(), db)
userDatabase := controller.NewUserDatabase(userDB, cache.NewUserCacheRedis(rdb, relation.NewUserGorm(db), cache.GetDefaultOpt()), tx.NewGorm(db))
groupDatabase := controller.InitGroupDatabase(db, rdb, mongo.GetDatabase())
conversationDatabase := controller.NewConversationDatabase(relation.NewConversationGorm(db), cache.NewConversationRedis(rdb, cache.GetDefaultOpt(), relation.NewConversationGorm(db)), tx.NewGorm(db))

1
pkg/common/cmd/rpc.go
View File

@ -27,7 +27,6 @@ func (a *RpcCmd) Exec() error {
}
func (a *RpcCmd) StartSvr(name string, rpcFn func(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error) error {
a.port = 10050
if a.GetPortFlag() == 0 {
return errors.New("port is required")
}

7
pkg/common/db/controller/chatlog.go
View File

@ -1,13 +1,14 @@
package controller
import (
"context"
relationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
)
type ChatLogDatabase interface {
CreateChatLog(msg *pbMsg.MsgDataToMQ) error
GetChatLog(chatLog *relationTb.ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []relationTb.ChatLogModel, error)
GetChatLog(ctx context.Context, chatLog *relationTb.ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []relationTb.ChatLogModel, error)
}
func NewChatLogDatabase(chatLogModelInterface relationTb.ChatLogModelInterface) ChatLogDatabase {
@ -22,6 +23,6 @@ func (c *chatLogDatabase) CreateChatLog(msg *pbMsg.MsgDataToMQ) error {
return c.chatLogModel.Create(msg)
}
func (c *chatLogDatabase) GetChatLog(chatLog *relationTb.ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []relationTb.ChatLogModel, error) {
return c.chatLogModel.GetChatLog(chatLog, pageNumber, showNumber, contentTypes)
func (c *chatLogDatabase) GetChatLog(ctx context.Context, chatLog *relationTb.ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []relationTb.ChatLogModel, error) {
return c.chatLogModel.GetChatLog(ctx, chatLog, pageNumber, showNumber, contentTypes)
}

29
pkg/common/db/controller/msg.go
View File

@ -2,7 +2,10 @@ package controller
import (
"fmt"
relation2 "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/relation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
"github.com/redis/go-redis/v9"
"gorm.io/gorm"
"time"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
@ -95,12 +98,14 @@ type CommonMsgDatabase interface {
RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error)
RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error)
SearchMessage(ctx context.Context, req *pbMsg.SearchMessageReq) (msgData []*sdkws.MsgData, err error)
GetChatLog(ctx context.Context, req *pbMsg.GetChatLogsReq, number int32, number2 int32, int32s []int32) (int64, []relation.ChatLogModel, error)
}
func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase {
func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel, msgMyqModel relation.ChatLogModelInterface) CommonMsgDatabase {
return &commonMsgDatabase{
msgDocDatabase: msgDocModel,
cache: cacheModel,
msgMyq: msgMyqModel,
producer: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.LatestMsgToRedis.Topic),
producerToMongo: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToMongo.Topic),
producerToPush: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToPush.Topic),
@ -108,10 +113,11 @@ func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheMo
}
}
func InitCommonMsgDatabase(rdb redis.UniversalClient, database *mongo.Database) CommonMsgDatabase {
func InitCommonMsgDatabase(rdb redis.UniversalClient, database *mongo.Database, dbGrom *gorm.DB) CommonMsgDatabase {
cacheModel := cache.NewMsgCacheModel(rdb)
msgDocModel := unrelation.NewMsgMongoDriver(database)
CommonMsgDatabase := NewCommonMsgDatabase(msgDocModel, cacheModel)
msgMyqModel := relation2.NewChatLogGorm(dbGrom)
CommonMsgDatabase := NewCommonMsgDatabase(msgDocModel, cacheModel, msgMyqModel)
return CommonMsgDatabase
}
@ -120,6 +126,7 @@ type commonMsgDatabase struct {
extendMsgDatabase unRelationTb.ExtendMsgSetModelInterface
extendMsgSetModel unRelationTb.ExtendMsgSetModel
msg unRelationTb.MsgDocModel
msgMyq relation.ChatLogModelInterface
cache cache.MsgModel
producer *kafka.Producer
producerToMongo *kafka.Producer
@ -127,6 +134,22 @@ type commonMsgDatabase struct {
producerToPush *kafka.Producer
}
func (db *commonMsgDatabase) GetChatLog(ctx context.Context, req *pbMsg.GetChatLogsReq, pageNumber int32, showNumber int32, int32s []int32) (int64, []relation.ChatLogModel, error) {
chatLog := &relation.ChatLogModel{
ContentType: req.ContentType,
RecvID: req.RecvID,
SendID: req.SendID,
}
if req.SendTime != "" {
sendTime, err := utils.TimeStringToTime(req.SendTime)
if err != nil {
return 0, nil, err
}
chatLog.SendTime = sendTime
}
return db.msgMyq.GetChatLog(ctx, chatLog, pageNumber, showNumber, int32s)
}
func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbMsg.SearchMessageReq) (msgData []*sdkws.MsgData, err error) {
var totalMsgs []*sdkws.MsgData
msgs, err := db.msgDocDatabase.SearchMessage(ctx, req)

19
pkg/common/db/relation/chat_log_model.go
View File

@ -1,6 +1,7 @@
package relation
import (
"context"
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
@ -48,30 +49,30 @@ func (c *ChatLogGorm) Create(msg *pbMsg.MsgDataToMQ) error {
return c.DB.Create(chatLog).Error
}
func (c *ChatLogGorm) GetChatLog(chatLog *relation.ChatLogModel, pageNumber, showNumber int32, contentTypeList []int32) (int64, []relation.ChatLogModel, error) {
func (c *ChatLogGorm) GetChatLog(ctx context.Context, chatLog *relation.ChatLogModel, pageNumber, showNumber int32, contentTypeList []int32) (int64, []relation.ChatLogModel, error) {
mdb := c.DB.Model(chatLog)
if chatLog.SendTime.Unix() > 0 {
mdb = mdb.Where("send_time > ? and send_time < ?", chatLog.SendTime, chatLog.SendTime.AddDate(0, 0, 1))
mdb = mdb.WithContext(ctx).Where("send_time > ? and send_time < ?", chatLog.SendTime, chatLog.SendTime.AddDate(0, 0, 1))
}
if chatLog.Content != "" {
mdb = mdb.Where(" content like ? ", fmt.Sprintf("%%%s%%", chatLog.Content))
mdb = mdb.WithContext(ctx).Where(" content like ? ", fmt.Sprintf("%%%s%%", chatLog.Content))
}
if chatLog.SessionType == 1 {
mdb = mdb.Where("session_type = ?", chatLog.SessionType)
mdb = mdb.WithContext(ctx).Where("session_type = ?", chatLog.SessionType)
} else if chatLog.SessionType == 2 {
mdb = mdb.Where("session_type in (?)", []int{constant.GroupChatType, constant.SuperGroupChatType})
mdb = mdb.WithContext(ctx).Where("session_type in (?)", []int{constant.GroupChatType, constant.SuperGroupChatType})
}
if chatLog.ContentType != 0 {
mdb = mdb.Where("content_type = ?", chatLog.ContentType)
mdb = mdb.WithContext(ctx).Where("content_type = ?", chatLog.ContentType)
}
if chatLog.SendID != "" {
mdb = mdb.Where("send_id = ?", chatLog.SendID)
mdb = mdb.WithContext(ctx).Where("send_id = ?", chatLog.SendID)
}
if chatLog.RecvID != "" {
mdb = mdb.Where("recv_id = ?", chatLog.RecvID)
mdb = mdb.WithContext(ctx).Where("recv_id = ?", chatLog.RecvID)
}
if len(contentTypeList) > 0 {
mdb = mdb.Where("content_type in (?)", contentTypeList)
mdb = mdb.WithContext(ctx).Where("content_type in (?)", contentTypeList)
}
var count int64
if err := mdb.Count(&count).Error; err != nil {

5
pkg/common/db/relation/group_model.go
View File

@ -2,6 +2,7 @@ package relation
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/ormutil"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
@ -46,7 +47,9 @@ func (g *GroupGorm) Take(ctx context.Context, groupID string) (group *relation.G
}
func (g *GroupGorm) Search(ctx context.Context, keyword string, pageNumber, showNumber int32) (total uint32, groups []*relation.GroupModel, err error) {
return ormutil.GormSearch[relation.GroupModel](g.DB, []string{"name"}, keyword, pageNumber, showNumber)
db := g.DB
db = db.WithContext(ctx).Where("status!=?", constant.GroupStatusDismissed)
return ormutil.GormSearch[relation.GroupModel](db, []string{"name"}, keyword, pageNumber, showNumber)
}
func (g *GroupGorm) GetGroupIDsByGroupType(ctx context.Context, groupType int) (groupIDs []string, err error) {

3
pkg/common/db/table/relation/chatlog.go
View File

@ -1,6 +1,7 @@
package relation
import (
"context"
"time"
pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
@ -34,5 +35,5 @@ func (ChatLogModel) TableName() string {
type ChatLogModelInterface interface {
Create(msg *pbMsg.MsgDataToMQ) error
GetChatLog(chatLog *ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []ChatLogModel, error)
GetChatLog(ctx context.Context, chatLog *ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []ChatLogModel, error)
}

966
pkg/proto/msg/msg.pb.go
View File

File diff suppressed because it is too large Load Diff

36
pkg/proto/msg/msg.proto
View File

@ -318,6 +318,41 @@ message SearchMessageResp{
repeated sdkws.MsgData msgs=1;
}
message ChatLog {
string serverMsgID = 1;
string clientMsgID = 2;
string sendID = 3;
string recvID = 4;
string groupID = 5;
string recvNickname = 6;
int32 senderPlatformID = 7;
string senderNickname = 8;
string senderFaceURL = 9;
string groupName = 10;
int32 sessionType = 11;
int32 msgFrom = 12;
int32 contentType = 13;
string content = 14;
int32 status = 15;
int64 sendTime = 16;
int64 createTime = 17;
string ex = 18;
}
message GetChatLogsReq{
string SendID=1;//发送者ID
string RecvID=2;//接收者ID
int32 contentType = 3;
string sendTime=4;
sdkws.RequestPagination pagination = 6;
}
message GetChatLogsResp{
repeated ChatLog chatLogs=1;
int32 chatLogsNum = 3;
}
service msg {
//获取最小最大seq（包括用户的，以及指定群组的）
rpc GetMaxSeq(sdkws.GetMaxSeqReq) returns(sdkws.GetMaxSeqResp);
@ -325,6 +360,7 @@ service msg {
//拉取历史消息（包括用户的，以及指定群组的）
rpc PullMessageBySeqs(sdkws.PullMessageBySeqsReq) returns(sdkws.PullMessageBySeqsResp);
rpc SearchMessage(SearchMessageReq) returns(SearchMessageResp);
rpc GetChatLogs(GetChatLogsReq)returns(GetChatLogsResp);
//发送消息
rpc SendMsg(SendMsgReq) returns(SendMsgResp);