diff --git a/build.cmd b/build.cmd deleted file mode 100644 index 4c53576c0..000000000 --- a/build.cmd +++ /dev/null @@ -1 +0,0 @@ -go build -o \ No newline at end of file diff --git a/docker-compose.yaml b/docker-compose.yaml index 0b32d2d50..1f7d09f82 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -100,7 +100,7 @@ services: openim_server: - image: ghcr.io/openimsdk/openim-server:v3.0.1 + image: ghcr.io/openimsdk/openim-server:v3.0.0 container_name: openim-server volumes: - ./logs:/Open-IM-Server/logs @@ -123,7 +123,7 @@ services: max-file: "2" openim_chat: - image: openim/openim_chat:v1.1.0 + image: ghcr.io/openimsdk/openim-chat:v1.0.0 container_name: openim_chat restart: always depends_on: diff --git a/init_docker.sh b/init_docker.sh old mode 100644 new mode 100755 index 09213bf50..9815fe871 --- a/init_docker.sh +++ b/init_docker.sh @@ -21,7 +21,12 @@ if ! command -v docker >/dev/null 2>&1; then fi # Start Docker services using docker-compose -docker-compose up -d +if command -v docker-compose &> /dev/null +then + docker-compose up -d +else + docker compose up -d +fi # Move back to the 'scripts' folder cd scripts diff --git a/install_guide.sh b/install_guide.sh old mode 100644 new mode 100755 diff --git a/install_im_compose.sh b/install_im_compose.sh old mode 100644 new mode 100755 diff --git a/install_im_server.sh b/install_im_server.sh old mode 100644 new mode 100755 diff --git a/internal/api/msg.go b/internal/api/msg.go index d8a3c292b..8ad7943e9 100644 --- a/internal/api/msg.go +++ b/internal/api/msg.go @@ -214,6 +214,7 @@ func (m *MessageApi) SendMessage(c *gin.Context) { if err != nil { log.ZError(c, "decodeData failed", err) apiresp.GinError(c, err) + return } sendMsgReq.MsgData.RecvID = req.RecvID var status int @@ -260,6 +261,7 @@ func (m *MessageApi) BatchSendMsg(c *gin.Context) { if err != nil { log.ZError(c, "GetAllUserIDs failed", err) apiresp.GinError(c, err) + return } if len(recvIDsPart) < showNumber { recvIDs = append(recvIDs, recvIDsPart...) @@ -275,6 +277,7 @@ func (m *MessageApi) BatchSendMsg(c *gin.Context) { if err != nil { log.ZError(c, "decodeData failed", err) apiresp.GinError(c, err) + return } for _, recvID := range recvIDs { sendMsgReq.MsgData.RecvID = recvID diff --git a/internal/api/user.go b/internal/api/user.go index 28cb74509..baea8fed3 100644 --- a/internal/api/user.go +++ b/internal/api/user.go @@ -83,7 +83,7 @@ func (u *UserApi) GetUsersOnlineStatus(c *gin.Context) { msgClient := msggateway.NewMsgGatewayClient(v) reply, err := msgClient.GetUsersOnlineStatus(c, &req) if err != nil { - log.ZWarn(c, "GetUsersOnlineStatus rpc err", err) + log.ZWarn(c, "GetUsersOnlineStatus rpc err", err) continue } else { wsResult = append(wsResult, reply.SuccessResult...) diff --git a/internal/msggateway/hub_server.go b/internal/msggateway/hub_server.go index 81aade1cb..6d852515c 100644 --- a/internal/msggateway/hub_server.go +++ b/internal/msggateway/hub_server.go @@ -25,9 +25,7 @@ import ( "github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/log" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome" - "github.com/OpenIMSDK/Open-IM-Server/pkg/common/tokenverify" "github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry" - "github.com/OpenIMSDK/Open-IM-Server/pkg/errs" "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msggateway" "github.com/OpenIMSDK/Open-IM-Server/pkg/startrpc" "github.com/OpenIMSDK/Open-IM-Server/pkg/utils" @@ -84,9 +82,6 @@ func (s *Server) GetUsersOnlineStatus( ctx context.Context, req *msggateway.GetUsersOnlineStatusReq, ) (*msggateway.GetUsersOnlineStatusResp, error) { - if !tokenverify.IsAppManagerUid(ctx) { - return nil, errs.ErrNoPermission.Wrap("only app manager") - } var resp msggateway.GetUsersOnlineStatusResp for _, userID := range req.UserIDs { clients, ok := s.LongConnServer.GetUserAllCons(userID) @@ -181,13 +176,12 @@ func (s *Server) KickUserOffline( if clients, _, ok := s.LongConnServer.GetUserPlatformCons(v, int(req.PlatformID)); ok { for _, client := range clients { log.ZDebug(ctx, "kick user offline", "userID", v, "platformID", req.PlatformID, "client", client) - err := client.KickOnlineMessage() - if err != nil { - return nil, err + if err := client.longConnServer.KickUserConn(client); err != nil { + log.ZWarn(ctx, "kick user offline failed", err, "userID", v, "platformID", req.PlatformID) } } } else { - log.ZWarn(ctx, "conn not exist", nil, "userID", v, "platformID", req.PlatformID) + log.ZInfo(ctx, "conn not exist", "userID", v, "platformID", req.PlatformID) } } return &msggateway.KickUserOfflineResp{}, nil diff --git a/internal/msggateway/n_ws_server.go b/internal/msggateway/n_ws_server.go index 2988e2c08..c2b46cf1f 100644 --- a/internal/msggateway/n_ws_server.go +++ b/internal/msggateway/n_ws_server.go @@ -47,6 +47,7 @@ type LongConnServer interface { Validate(s interface{}) error SetCacheHandler(cache cache.MsgModel) SetDiscoveryRegistry(client discoveryregistry.SvcDiscoveryRegistry) + KickUserConn(client *Client) error UnRegister(c *Client) Compressor Encoder @@ -145,7 +146,7 @@ func (ws *WsServer) Run() error { case client = <-ws.unregisterChan: ws.unregisterClient(client) case onlineInfo := <-ws.kickHandlerChan: - ws.multiTerminalLoginChecker(onlineInfo) + ws.multiTerminalLoginChecker(onlineInfo.clientOK, onlineInfo.oldClients, onlineInfo.newClient) } } }() @@ -207,80 +208,77 @@ func getRemoteAdders(client []*Client) string { return ret } -func (ws *WsServer) multiTerminalLoginChecker(info *kickHandler) { +func (ws *WsServer) KickUserConn(client *Client) error { + ws.clients.deleteClients(client.UserID, []*Client{client}) + return client.KickOnlineMessage() +} + +func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Client, newClient *Client) { switch config.Config.MultiLoginPolicy { case constant.DefalutNotKick: case constant.PCAndOther: - if constant.PlatformIDToClass(info.newClient.PlatformID) == constant.TerminalPC { + if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC { return } fallthrough case constant.AllLoginButSameTermKick: - if info.clientOK { - ws.clients.deleteClients(info.newClient.UserID, info.oldClients) - for _, c := range info.oldClients { + if clientOK { + ws.clients.deleteClients(newClient.UserID, oldClients) + for _, c := range oldClients { err := c.KickOnlineMessage() if err != nil { log.ZWarn(c.ctx, "KickOnlineMessage", err) } } m, err := ws.cache.GetTokensWithoutError( - info.newClient.ctx, - info.newClient.UserID, - info.newClient.PlatformID, + newClient.ctx, + newClient.UserID, + newClient.PlatformID, ) if err != nil && err != redis.Nil { log.ZWarn( - info.newClient.ctx, + newClient.ctx, "get token from redis err", err, "userID", - info.newClient.UserID, + newClient.UserID, "platformID", - info.newClient.PlatformID, + newClient.PlatformID, ) return } if m == nil { log.ZWarn( - info.newClient.ctx, + newClient.ctx, "m is nil", errors.New("m is nil"), "userID", - info.newClient.UserID, + newClient.UserID, "platformID", - info.newClient.PlatformID, + newClient.PlatformID, ) return } log.ZDebug( - info.newClient.ctx, + newClient.ctx, "get token from redis", "userID", - info.newClient.UserID, + newClient.UserID, "platformID", - info.newClient.PlatformID, + newClient.PlatformID, "tokenMap", m, ) for k := range m { - if k != info.newClient.ctx.GetToken() { + if k != newClient.ctx.GetToken() { m[k] = constant.KickedToken } } - log.ZDebug(info.newClient.ctx, "set token map is ", "token map", m, "userID", info.newClient.UserID) - err = ws.cache.SetTokenMapByUidPid(info.newClient.ctx, info.newClient.UserID, info.newClient.PlatformID, m) + log.ZDebug(newClient.ctx, "set token map is ", "token map", m, "userID", newClient.UserID) + err = ws.cache.SetTokenMapByUidPid(newClient.ctx, newClient.UserID, newClient.PlatformID, m) if err != nil { - log.ZWarn( - info.newClient.ctx, - "SetTokenMapByUidPid err", - err, - "userID", - info.newClient.UserID, - "platformID", - info.newClient.PlatformID, - ) + log.ZWarn(newClient.ctx, "SetTokenMapByUidPid err", err, "userID", newClient.UserID, "platformID", newClient.PlatformID) return } } diff --git a/internal/rpc/auth/auth.go b/internal/rpc/auth/auth.go index 48e21fbce..317754743 100644 --- a/internal/rpc/auth/auth.go +++ b/internal/rpc/auth/auth.go @@ -23,6 +23,7 @@ import ( "github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller" + "github.com/OpenIMSDK/Open-IM-Server/pkg/common/log" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/tokenverify" "github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry" @@ -129,11 +130,16 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID if err != nil { return err } + for _, v := range conns { + log.ZDebug(ctx, "forceKickOff", "conn", v.(*grpc.ClientConn).Target()) + } for _, v := range conns { client := msggateway.NewMsgGatewayClient(v) kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID} _, err := client.KickUserOffline(ctx, kickReq) - return utils.Wrap(err, "") + if err != nil { + log.ZError(ctx, "forceKickOff", err, "kickReq", kickReq) + } } return nil } diff --git a/internal/rpc/msg/sync_msg.go b/internal/rpc/msg/sync_msg.go index 8cfac93bd..70967a1ba 100644 --- a/internal/rpc/msg/sync_msg.go +++ b/internal/rpc/msg/sync_msg.go @@ -108,11 +108,11 @@ func (m *msgServer) GetMaxSeq(ctx context.Context, req *sdkws.GetMaxSeqReq) (*sd func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq) (resp *msg.SearchMessageResp, err error) { var chatLogs []*sdkws.MsgData + var total int32 resp = &msg.SearchMessageResp{} - if chatLogs, err = m.MsgDatabase.SearchMessage(ctx, req); err != nil { + if total, chatLogs, err = m.MsgDatabase.SearchMessage(ctx, req); err != nil { return nil, err } - var num int for _, chatLog := range chatLogs { pbChatLog := &msg.ChatLog{} utils.CopyStructFields(pbChatLog, chatLog) @@ -146,9 +146,8 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq pbChatLog.GroupType = group.GroupType } resp.ChatLogs = append(resp.ChatLogs, pbChatLog) - num++ } - resp.ChatLogsNum = int32(num) + resp.ChatLogsNum = total return resp, nil } diff --git a/internal/tools/cron_task.go b/internal/tools/cron_task.go index 5e4183615..82343157b 100644 --- a/internal/tools/cron_task.go +++ b/internal/tools/cron_task.go @@ -41,11 +41,11 @@ func StartCronTask() error { panic(err) } log.ZInfo(context.Background(), "start msgDestruct cron task", "cron config", config.Config.MsgDestructTime) - _, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs) - if err != nil { - fmt.Println("start conversationsDestructMsgs cron failed", err.Error(), config.Config.ChatRecordsClearTime) - panic(err) - } + // _, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs) + // if err != nil { + // fmt.Println("start conversationsDestructMsgs cron failed", err.Error(), config.Config.ChatRecordsClearTime) + // panic(err) + // } c.Start() wg.Wait() return nil diff --git a/pkg/common/db/cache/conversation.go b/pkg/common/db/cache/conversation.go index d3a9d44aa..119967c70 100644 --- a/pkg/common/db/cache/conversation.go +++ b/pkg/common/db/cache/conversation.go @@ -54,7 +54,7 @@ type ConversationCache interface { // get one conversation from msgCache GetConversation(ctx context.Context, ownerUserID, conversationID string) (*relationTb.ConversationModel, error) - DelConvsersations(ownerUserID string, conversationIDs ...string) ConversationCache + DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache // get one conversation from msgCache GetConversations( @@ -225,9 +225,9 @@ func (c *ConversationRedisCache) GetConversation( ) } -func (c *ConversationRedisCache) DelConvsersations(ownerUserID string, convsersationIDs ...string) ConversationCache { +func (c *ConversationRedisCache) DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache { var keys []string - for _, conversationID := range convsersationIDs { + for _, conversationID := range conversationIDs { keys = append(keys, c.getConversationKey(ownerUserID, conversationID)) } cache := c.NewCache() diff --git a/pkg/common/db/controller/conversation.go b/pkg/common/db/controller/conversation.go index 3cb4d1055..d65315918 100644 --- a/pkg/common/db/controller/conversation.go +++ b/pkg/common/db/controller/conversation.go @@ -104,7 +104,7 @@ func (c *conversationDatabase) SetUsersConversationFiledTx(ctx context.Context, if err != nil { return err } - cache = cache.DelConversationIDs(NotUserIDs...).DelUserConversationIDsHash(NotUserIDs...).DelConvsersations(conversation.ConversationID, NotUserIDs...) + cache = cache.DelConversationIDs(NotUserIDs...).DelUserConversationIDsHash(NotUserIDs...).DelConversations(conversation.ConversationID, NotUserIDs...) } return nil }); err != nil { @@ -128,7 +128,7 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat var userIDs []string cache := c.cache.NewCache() for _, conversation := range conversations { - cache = cache.DelConvsersations(conversation.OwnerUserID, conversation.ConversationID) + cache = cache.DelConversations(conversation.OwnerUserID, conversation.ConversationID) userIDs = append(userIDs, conversation.OwnerUserID) } return cache.DelConversationIDs(userIDs...).DelUserConversationIDsHash(userIDs...).ExecDel(ctx) @@ -190,7 +190,7 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs var conversationIDs []string for _, conversation := range conversations { conversationIDs = append(conversationIDs, conversation.ConversationID) - cache = cache.DelConvsersations(conversation.OwnerUserID, conversation.ConversationID) + cache = cache.DelConversations(conversation.OwnerUserID, conversation.ConversationID) } conversationTx := c.conversationDB.NewTx(tx) existConversations, err := conversationTx.Find(ctx, ownerUserID, conversationIDs) @@ -247,7 +247,7 @@ func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context, for _, v := range notExistUserIDs { conversation := relationTb.ConversationModel{ConversationType: constant.SuperGroupChatType, GroupID: groupID, OwnerUserID: v, ConversationID: conversationID} conversations = append(conversations, &conversation) - cache = cache.DelConvsersations(v, conversationID) + cache = cache.DelConversations(v, conversationID) } cache = cache.DelConversationIDs(notExistUserIDs...).DelUserConversationIDsHash(notExistUserIDs...) if len(conversations) > 0 { @@ -261,7 +261,7 @@ func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context, return err } for _, v := range existConversationUserIDs { - cache = cache.DelConvsersations(v, conversationID) + cache = cache.DelConversations(v, conversationID) } return nil }); err != nil { diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index fdda23ac9..37a71bafa 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -92,7 +92,7 @@ type CommonMsgDatabase interface { GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) SetSendMsgStatus(ctx context.Context, id string, status int32) error GetSendMsgStatus(ctx context.Context, id string) (int32, error) - SearchMessage(ctx context.Context, req *pbMsg.SearchMessageReq) (msgData []*sdkws.MsgData, err error) + SearchMessage(ctx context.Context, req *pbMsg.SearchMessageReq) (total int32, msgData []*sdkws.MsgData, err error) // to mq MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error @@ -940,14 +940,14 @@ func (db *commonMsgDatabase) RangeGroupSendCount( return db.msgDocDatabase.RangeGroupSendCount(ctx, start, end, ase, pageNumber, showNumber) } -func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbMsg.SearchMessageReq) (msgData []*sdkws.MsgData, err error) { +func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbMsg.SearchMessageReq) (total int32, msgData []*sdkws.MsgData, err error) { var totalMsgs []*sdkws.MsgData - msgs, err := db.msgDocDatabase.SearchMessage(ctx, req) + total, msgs, err := db.msgDocDatabase.SearchMessage(ctx, req) if err != nil { - return nil, err + return 0, nil, err } for _, msg := range msgs { totalMsgs = append(totalMsgs, convert.MsgDB2Pb(msg.Msg)) } - return totalMsgs, nil + return total, totalMsgs, nil } diff --git a/pkg/common/db/table/unrelation/msg.go b/pkg/common/db/table/unrelation/msg.go index ac4602d35..dc16613ed 100644 --- a/pkg/common/db/table/unrelation/msg.go +++ b/pkg/common/db/table/unrelation/msg.go @@ -110,7 +110,7 @@ type MsgDocModelInterface interface { GetMsgDocModelByIndex(ctx context.Context, conversationID string, index, sort int64) (*MsgDocModel, error) DeleteMsgsInOneDocByIndex(ctx context.Context, docID string, indexes []int) error MarkSingleChatMsgsAsRead(ctx context.Context, userID string, docID string, indexes []int64) error - SearchMessage(ctx context.Context, req *msg.SearchMessageReq) ([]*MsgInfoModel, error) + SearchMessage(ctx context.Context, req *msg.SearchMessageReq) (int32, []*MsgInfoModel, error) RangeUserSendCount( ctx context.Context, start time.Time, diff --git a/pkg/common/db/unrelation/msg.go b/pkg/common/db/unrelation/msg.go index 222b70190..b7403ebde 100644 --- a/pkg/common/db/unrelation/msg.go +++ b/pkg/common/db/unrelation/msg.go @@ -1067,20 +1067,20 @@ func (m *MsgMongoDriver) RangeGroupSendCount( return result[0].MsgCount, result[0].UserCount, groups, dateCount, nil } -func (m *MsgMongoDriver) SearchMessage(ctx context.Context, req *msg.SearchMessageReq) ([]*table.MsgInfoModel, error) { - msgs, err := m.searchMessage(ctx, req) +func (m *MsgMongoDriver) SearchMessage(ctx context.Context, req *msg.SearchMessageReq) (int32, []*table.MsgInfoModel, error) { + total, msgs, err := m.searchMessage(ctx, req) if err != nil { - return nil, err + return 0, nil, err } for _, msg1 := range msgs { if msg1.IsRead { msg1.Msg.IsRead = true } } - return msgs, nil + return total, msgs, nil } -func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessageReq) ([]*table.MsgInfoModel, error) { +func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessageReq) (int32, []*table.MsgInfoModel, error) { var pipe mongo.Pipeline condition := bson.A{} if req.SendTime != "" { @@ -1153,16 +1153,16 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa } cursor, err := m.MsgCollection.Aggregate(ctx, pipe) if err != nil { - return nil, err + return 0, nil, err } var msgsDocs []table.MsgDocModel err = cursor.All(ctx, &msgsDocs) if err != nil { - return nil, err + return 0, nil, err } if len(msgsDocs) == 0 { - return nil, errs.Wrap(mongo.ErrNoDocuments) + return 0, nil, errs.Wrap(mongo.ErrNoDocuments) } msgs := make([]*table.MsgInfoModel, 0) for index := range msgsDocs { @@ -1187,14 +1187,14 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa } data, err := json.Marshal(&revokeContent) if err != nil { - return nil, err + return 0, nil, err } elem := sdkws.NotificationElem{ Detail: string(data), } content, err := json.Marshal(&elem) if err != nil { - return nil, err + return 0, nil, err } msg.Msg.ContentType = constant.MsgRevokeNotification msg.Msg.Content = string(content) @@ -1209,5 +1209,5 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa } else { msgs = msgs[start:] } - return msgs, nil + return n, msgs, nil } diff --git a/pkg/proto/msg/msg.go b/pkg/proto/msg/msg.go index 62b776856..03c9da915 100644 --- a/pkg/proto/msg/msg.go +++ b/pkg/proto/msg/msg.go @@ -126,6 +126,11 @@ func (x *MarkMsgsAsReadReq) Check() error { if x.UserID == "" { return errs.ErrArgs.Wrap("userID is empty") } + for _, seq := range x.Seqs { + if seq == 0 { + return errs.ErrArgs.Wrap("seqs has 0 value is invalid") + } + } return nil } @@ -139,6 +144,11 @@ func (x *MarkConversationAsReadReq) Check() error { if x.HasReadSeq < 1 { return errs.ErrArgs.Wrap("hasReadSeq is invalid") } + for _, seq := range x.Seqs { + if seq == 0 { + return errs.ErrArgs.Wrap("seqs has 0 value is invalid") + } + } return nil }