diff --git a/README.md b/README.md index b180f4438..dcf7230c1 100644 --- a/README.md +++ b/README.md @@ -182,7 +182,7 @@ We take notes of each [biweekly meeting](https://github.com/OpenIMSDK/Open-IM-Se ## Who are using Open-IM-Server The [user case studies](https://github.com/OpenIMSDK/community/blob/main/ADOPTERS.md) page includes the user list of the project. You can leave a [📝comment](https://github.com/OpenIMSDK/Open-IM-Server/issues/379) to let us know your use case. -![avatar](https://github.com/OpenIMSDK/OpenIM-Docs/blob/main/docs/images/WechatIMG20.jpeg) +![avatar](https://openim-1253691595.cos.ap-nanjing.myqcloud.com/WechatIMG20.jpeg) ## License diff --git a/docker-compose.yaml b/docker-compose.yaml index 0b32d2d50..23565c23a 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 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/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/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 } diff --git a/start.bat b/start.bat index 2b5288796..6ffee0882 100644 --- a/start.bat +++ b/start.bat @@ -8,5 +8,5 @@ start msg.exe -p 10030 start msggateway.exe -p 10040 -w 10001 start push.exe -p 10070 start msgtransfer.exe -start third.exe -p 10090 start user.exe -p 10010 +start third.exe -p 10090