diff --git a/.github/workflows/openimci.yml b/.github/workflows/openimci.yml index d8e988f0b..d16e91ba4 100644 --- a/.github/workflows/openimci.yml +++ b/.github/workflows/openimci.yml @@ -201,7 +201,6 @@ jobs: - name: Build, Start, Check Services and Print Logs for Ubuntu if: runner.os == 'Linux' run: | - sudo make init && \ sudo make build && \ sudo make start && \ sudo make check || \ diff --git a/build/goreleaser.yaml b/build/goreleaser.yaml index ed7f7cd1b..93fe9f4c8 100644 --- a/build/goreleaser.yaml +++ b/build/goreleaser.yaml @@ -3,12 +3,36 @@ before: hooks: + - make clean # You may remove this if you don't use go modules. - make tidy - make copyright.add # you may remove this if you don't need go generate - go generate ./... +git: + # What should be used to sort tags when gathering the current and previous + # tags if there are more than one tag in the same commit. + # + # Default: '-version:refname' + tag_sort: -version:creatordate + + # What should be used to specify prerelease suffix while sorting tags when gathering + # the current and previous tags if there are more than one tag in the same commit. + # + # Since: v1.17 + prerelease_suffix: "-" + + # Tags to be ignored by GoReleaser. + # This means that GoReleaser will not pick up tags that match any of the + # provided values as either previous or current tags. + # + # Templates: allowed. + # Since: v1.21. + ignore_tags: + - nightly + # - "{{.Env.IGNORE_TAG}}" + snapshot: name_template: "{{ incpatch .Version }}-next" @@ -495,4 +519,4 @@ checksum: algorithm: sha256 release: - prerelease: auto \ No newline at end of file + prerelease: auto diff --git a/config/templates/config.yaml.template b/config/templates/config.yaml.template index fd51a2e31..9954a8863 100644 --- a/config/templates/config.yaml.template +++ b/config/templates/config.yaml.template @@ -312,6 +312,14 @@ callback: enable: false timeout: 5 failedContinue: true + beforeUpdateUserInfoEx: + enable: false + timeout: 5 + failedContinue: true + afterUpdateUserInfoEx: + enable: false + timeout: 5 + failedContinue: true afterSendSingleMsg: enable: false timeout: 5 @@ -498,7 +506,7 @@ callback: # The number of ports needs to be consistent with msg_transfer_service_num in script/path_info.sh prometheus: enable: false - prometheusUrl: 172.28.0.1:13000 + grafanaUrl: 172.28.0.1:13000 apiPrometheusPort: [20100] userPrometheusPort: [ 20110 ] friendPrometheusPort: [ 20120 ] diff --git a/deployments/templates/env-template.yaml b/deployments/templates/env-template.yaml index ab9c87c02..e0fd7bc9f 100644 --- a/deployments/templates/env-template.yaml +++ b/deployments/templates/env-template.yaml @@ -1,42 +1,23 @@ -# Copyright © 2023 OpenIM. All rights reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# ====================================== -# ========= Basic Configuration ======== -# ====================================== - -# The user for authentication or system operations. -# Default: OPENIM_USER=root -USER=${OPENIM_USER} - -# Password associated with the specified user for authentication. -# Default: PASSWORD=openIM123 -PASSWORD=${PASSWORD} - -# Base URL for the application programming interface (API). -# Default: API_URL=http://172.28.0.1:10002 -API_URL=${API_URL} - -# Directory path for storing data files or related information. -# Default: DATA_DIR=./ +# ----------------------------------------------------------------------------- +# General Configuration +# This section contains general configuration options for the entire environment. +# These options can be set via environment variables. If both environment variables +# and settings in this .env file exist, the environment variables take precedence. +# ----------------------------------------------------------------------------- +# ========================== +# General Configuration +# ========================== +# These settings apply to the overall environment. + +# Data storage directory for persistent data. +# Example: DATA_DIR=/path/to/data DATA_DIR=${DATA_DIR} -# Choose the appropriate image address, the default is GITHUB image, -# you can choose docker hub, for Chinese users can choose Ali Cloud -# export IMAGE_REGISTRY="ghcr.io/openimsdk" -# export IMAGE_REGISTRY="openim" -# export IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk" +# Docker image registry. Uncomment the preferred one. +# Options: ghcr.io/openimsdk, openim, registry.cn-hangzhou.aliyuncs.com/openimsdk +# IMAGE_REGISTRY="ghcr.io/openimsdk" +# IMAGE_REGISTRY="openim" +# IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk" IMAGE_REGISTRY=${IMAGE_REGISTRY} # ====================================== @@ -47,10 +28,9 @@ IMAGE_REGISTRY=${IMAGE_REGISTRY} # Default: DOCKER_BRIDGE_SUBNET=172.28.0.0/16 DOCKER_BRIDGE_SUBNET=${DOCKER_BRIDGE_SUBNET} -# Gateway for the Docker network. -# Default: DOCKER_BRIDGE_GATEWAY=172.28.0.1 +# Set and specify the IP addresses of some containers. Generally speaking, +# you do not need to modify these configurations to facilitate debugging DOCKER_BRIDGE_GATEWAY=${DOCKER_BRIDGE_GATEWAY} - MONGO_NETWORK_ADDRESS=${MONGO_NETWORK_ADDRESS} REDIS_NETWORK_ADDRESS=${REDIS_NETWORK_ADDRESS} KAFKA_NETWORK_ADDRESS=${KAFKA_NETWORK_ADDRESS} @@ -65,25 +45,45 @@ NODE_EXPORTER_NETWORK_ADDRESS=${NODE_EXPORTER_NETWORK_ADDRESS} OPENIM_ADMIN_FRONT_NETWORK_ADDRESS=${OPENIM_ADMIN_FRONT_NETWORK_ADDRESS} ALERT_MANAGER_NETWORK_ADDRESS=${ALERT_MANAGER_NETWORK_ADDRESS} -# =============================================== -# = Component Extension Configuration = -# =============================================== +# ============================================================================== +# Configuration Update Instructions +# ============================================================================== +# This header outlines the methods to update common variables in config.yaml and .env files. +# These instructions are vital for maintaining the OpenIM environment's configuration. +# +# METHOD 1: Regenerate All Configurations +# ---------------------------------------- +# Use this method to regenerate all configurations. +# Steps: +# 1. Delete existing config files: +# - openim-server/config/config.yaml +# - openim-chat/config/config.yaml +# 2. Modify the .env file as required. +# 3. Run 'docker compose up -d'. This will regenerate: +# - config/config.yaml +# +# METHOD 2: Modify Individual Configuration Files +# ----------------------------------------------- +# Use this method to update specific configuration files. +# Steps: +# 1. Modify the .env file as necessary. +# 2. Update the corresponding entries in: +# - config/config.yaml +# 3. Restart the services with 'docker compose up -d'. +# 4. Special Note: If you modify OPENIM_IP, API_OPENIM_PORT, or MINIO_PORT in .env, +# ensure to update the corresponding services and configurations accordingly. +# +# It is essential to follow these methods to ensure consistent and correct application behavior. +# ============================================================================== +# Local IP address of the service. Modify if necessary. +# Example: OPENIM_IP=172.28.0.1, +OPENIM_IP=${OPENIM_IP} -# ============ Component Extension Configuration ========== # ----- ZooKeeper Configuration ----- -# Address or hostname for the ZooKeeper service. -# Default: ZOOKEEPER_ADDRESS=172.28.0.1 -ZOOKEEPER_ADDRESS=${ZOOKEEPER_NETWORK_ADDRESS} - # Port for ZooKeeper service. # Default: ZOOKEEPER_PORT=12181 ZOOKEEPER_PORT=${ZOOKEEPER_PORT} -# ----- MongoDB Configuration ----- -# Address or hostname for the MongoDB service. -# Default: MONGO_ADDRESS=172.28.0.1 -MONGO_ADDRESS=${MONGO_NETWORK_ADDRESS} - # Port on which MongoDB service is running. # Default: MONGO_PORT=37017 # MONGO_PORT=${MONGO_PORT} @@ -101,9 +101,6 @@ MONGO_PASSWORD=${MONGO_PASSWORD} MONGO_DATABASE=${MONGO_DATABASE} # ----- Redis Configuration ----- -# Address or hostname for the Redis service. -# Default: REDIS_ADDRESS=172.28.0.1 -REDIS_ADDRESS=${REDIS_NETWORK_ADDRESS} # Port on which Redis in-memory data structure store is running. # Default: REDIS_PORT=16379 @@ -113,11 +110,6 @@ REDIS_PORT=${REDIS_PORT} # Default: REDIS_PASSWORD=openIM123 REDIS_PASSWORD=${REDIS_PASSWORD} -# ----- Kafka Configuration ----- -# Address or hostname for the Kafka service. -# Default: KAFKA_ADDRESS=172.28.0.1 -KAFKA_ADDRESS=${KAFKA_NETWORK_ADDRESS} - # Kakfa username to authenticate with the Kafka service. # KAFKA_USERNAME=${KAFKA_USERNAME} @@ -129,20 +121,13 @@ KAFKA_PORT=${KAFKA_PORT} # Default: KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis KAFKA_LATESTMSG_REDIS_TOPIC=${KAFKA_LATESTMSG_REDIS_TOPIC} -# Topic in Kafka for pushing messages (e.g. notifications or updates). -# Default: KAFKA_MSG_PUSH_TOPIC=msgToPush -KAFKA_MSG_PUSH_TOPIC=${KAFKA_MSG_PUSH_TOPIC} - -# Topic in Kafka for storing offline messages in MongoDB. -# Default: KAFKA_OFFLINEMSG_MONGO_TOPIC=offlineMsgToMongoMysql -KAFKA_OFFLINEMSG_MONGO_TOPIC=${KAFKA_OFFLINEMSG_MONGO_TOPIC} - -# ----- MinIO Configuration ---- -# Address or hostname for the MinIO object storage service. -# Default: MINIO_ADDRESS=172.28.0.1 -MINIO_ADDRESS=${MINIO_NETWORK_ADDRESS} - -# Port on which MinIO object storage service is running. +# MINIO_PORT +# ---------- +# MINIO_PORT sets the port for the MinIO object storage service. +# Upon changing this port, the MinIO endpoint URLs in the `config/config.yaml` file must be updated +# to reflect this change. The endpoints include both the 'endpoint' and 'signEndpoint' +# under the MinIO configuration. +# # Default: MINIO_PORT=10005 MINIO_PORT=${MINIO_PORT} @@ -155,19 +140,11 @@ MINIO_PORT=${MINIO_PORT} MINIO_SECRET_KEY=${MINIO_SECRET_KEY} # ----- Prometheus Configuration ----- -# Address or hostname for the Prometheus service. -# Default: PROMETHEUS_ADDRESS=172.28.0.1 -PROMETHEUS_ADDRESS=${PROMETHEUS_NETWORK_ADDRESS} - # Port on which Prometheus service is running. # Default: PROMETHEUS_PORT=19090 PROMETHEUS_PORT=${PROMETHEUS_PORT} # ----- Grafana Configuration ----- -# Address or hostname for the Grafana service. -# Default: GRAFANA_ADDRESS=172.28.0.1 -GRAFANA_ADDRESS=${GRAFANA_NETWORK_ADDRESS} - # Port on which Grafana service is running. # Default: GRAFANA_PORT=13000 GRAFANA_PORT=${GRAFANA_PORT} @@ -184,23 +161,19 @@ OPENIM_WEB_DIST_PATH=${OPENIM_WEB_DIST_PATH} # Default: OPENIM_WEB_PORT=11001 OPENIM_WEB_PORT=${OPENIM_WEB_PORT} -# Address or hostname for the OpenIM web service. -# Default: OPENIM_WEB_ADDRESS=172.28.0.1 -OPENIM_WEB_ADDRESS=${OPENIM_WEB_NETWORK_ADDRESS} - # ====================================== # ========= OpenIM Server ============== # ====================================== - -# Address or hostname for the OpenIM server. -# Default: OPENIM_SERVER_ADDRESS=172.28.0.1 -OPENIM_SERVER_ADDRESS=${OPENIM_SERVER_NETWORK_ADDRESS} - # Port for the OpenIM WebSockets. # Default: OPENIM_WS_PORT=10001 OPENIM_WS_PORT=${OPENIM_WS_PORT} -# Port for the OpenIM API. +# API_OPENIM_PORT +# --------------- +# This variable defines the port on which the OpenIM API service will listen. +# When changing this port, it's essential to update the apiURL in the config.yaml file +# to ensure the API service is accessible at the new port. +# # Default: API_OPENIM_PORT=10002 API_OPENIM_PORT=${API_OPENIM_PORT} @@ -213,10 +186,6 @@ API_OPENIM_PORT=${API_OPENIM_PORT} # Default: CHAT_IMAGE_VERSION=main CHAT_IMAGE_VERSION=${CHAT_IMAGE_VERSION} -# Address or hostname for the OpenIM chat service. -# Default: OPENIM_CHAT_ADDRESS=172.28.0.1 -OPENIM_CHAT_ADDRESS=${OPENIM_CHAT_NETWORK_ADDRESS} - # Port for the OpenIM chat API. # Default: OPENIM_CHAT_API_PORT=10008 OPENIM_CHAT_API_PORT=${OPENIM_CHAT_API_PORT} @@ -225,7 +194,6 @@ OPENIM_CHAT_API_PORT=${OPENIM_CHAT_API_PORT} # Default: OPENIM_CHAT_DATA_DIR=./openim-chat/main OPENIM_CHAT_DATA_DIR=${OPENIM_CHAT_DATA_DIR} - # ====================================== # ========== OpenIM Admin ============== # ====================================== diff --git a/deployments/templates/openim.yaml b/deployments/templates/openim.yaml index cbb4a8c6e..96d867e10 100644 --- a/deployments/templates/openim.yaml +++ b/deployments/templates/openim.yaml @@ -506,7 +506,7 @@ callback: # The number of ports needs to be consistent with msg_transfer_service_num in script/path_info.sh prometheus: enable: ${PROMETHEUS_ENABLE} - prometheusUrl: ${PROMETHEUS_URL} + grafanaUrl: ${GRAFANA_URL} apiPrometheusPort: [${API_PROM_PORT}] userPrometheusPort: [ ${USER_PROM_PORT} ] friendPrometheusPort: [ ${FRIEND_PROM_PORT} ] diff --git a/docker-compose.yml b/docker-compose.yml index de3deaaea..fd71896a7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -122,9 +122,9 @@ services: server: ipv4_address: ${OPENIM_WEB_NETWORK_ADDRESS:-172.28.0.7} - # Uncomment and configure the following services as needed + ## Uncomment and configure the following services as needed # openim-admin: - # image: ${IMAGE_REGISTRY:-ghcr.io/openimsdk}/openim-admin-front:v3.4.0 + # image: ${IMAGE_REGISTRY:-ghcr.io/openimsdk}/openim-admin:toc-base-open-docker.35 # container_name: openim-admin # restart: always # ports: @@ -167,6 +167,12 @@ services: # hostname: grafana # user: root # restart: always + # environment: + # - GF_SECURITY_ALLOW_EMBEDDING=true + # - GF_SESSION_COOKIE_SAMESITE=none + # - GF_SESSION_COOKIE_SECURE=true + # - GF_AUTH_ANONYMOUS_ENABLED=true + # - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin # ports: # - "${GRAFANA_PORT:-13000}:3000" # volumes: diff --git a/go.mod b/go.mod index 8d3890fb1..82558601f 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.19 require ( firebase.google.com/go v3.13.0+incompatible - github.com/OpenIMSDK/protocol v0.0.40 + github.com/OpenIMSDK/protocol v0.0.43 github.com/OpenIMSDK/tools v0.0.21 github.com/bwmarrin/snowflake v0.3.0 // indirect github.com/dtm-labs/rockscache v0.1.1 @@ -157,4 +157,4 @@ require ( gopkg.in/ini.v1 v1.67.0 // indirect ) -replace github.com/OpenIMSDK/protocol v0.0.40 => github.com/luhaoling/protocol v0.0.0-20231228025636-9ddb53aee5a4 +replace github.com/OpenIMSDK/protocol v0.0.43 => github.com/luhaoling/protocol v0.0.0-20240104030145-7b42ed9c02e9 diff --git a/go.sum b/go.sum index 13f3bd400..2d25fd2b8 100644 --- a/go.sum +++ b/go.sum @@ -225,8 +225,8 @@ github.com/lestrrat-go/strftime v1.0.6 h1:CFGsDEt1pOpFNU+TJB0nhz9jl+K0hZSLE205Ah github.com/lestrrat-go/strftime v1.0.6/go.mod h1:f7jQKgV5nnJpYgdEasS+/y7EsTb8ykN2z68n3TtcTaw= github.com/lithammer/shortuuid v3.0.0+incompatible h1:NcD0xWW/MZYXEHa6ITy6kaXN5nwm/V115vj2YXfhS0w= github.com/lithammer/shortuuid v3.0.0+incompatible/go.mod h1:FR74pbAuElzOUuenUHTK2Tciko1/vKuIKS9dSkDrA4w= -github.com/luhaoling/protocol v0.0.0-20231228025636-9ddb53aee5a4 h1:p9NaRVR7ba+i56AJMaiuASVnYf7FlSGi1joGn4f29bY= -github.com/luhaoling/protocol v0.0.0-20231228025636-9ddb53aee5a4/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y= +github.com/luhaoling/protocol v0.0.0-20240104030145-7b42ed9c02e9 h1:MFqmTU3Y9Sjq7wyABcwu1fk1bJLgiekiTflTiQSCq4Q= +github.com/luhaoling/protocol v0.0.0-20240104030145-7b42ed9c02e9/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y= github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA= github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo= diff --git a/internal/api/friend.go b/internal/api/friend.go index 0baaf0e15..7dc898a02 100644 --- a/internal/api/friend.go +++ b/internal/api/friend.go @@ -92,7 +92,6 @@ func (o *FriendApi) GetFriendIDs(c *gin.Context) { func (o *FriendApi) GetSpecifiedFriendsInfo(c *gin.Context) { a2r.Call(friend.FriendClient.GetSpecifiedFriendsInfo, o.Client, c) } - -func (o *FriendApi) SetPinFriends(c *gin.Context) { +func (o *FriendApi) UpdateFriends(c *gin.Context) { a2r.Call(friend.FriendClient.UpdateFriends, o.Client, c) } diff --git a/internal/api/msg.go b/internal/api/msg.go index b356f38e1..548450534 100644 --- a/internal/api/msg.go +++ b/internal/api/msg.go @@ -172,6 +172,7 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM if err = m.userRpcClient.GetNotificationByID(c, req.SendID); err != nil { return nil, err } + default: return nil, errs.ErrArgs.WithDetail("not support err contentType") } @@ -185,39 +186,63 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM return m.newUserSendMsgReq(c, &req), nil } +// SendMessage handles the sending of a message. It's an HTTP handler function to be used with Gin framework. func (m *MessageApi) SendMessage(c *gin.Context) { + // Initialize a request struct for sending a message. req := apistruct.SendMsgReq{} + + // Bind the JSON request body to the request struct. if err := c.BindJSON(&req); err != nil { + // Respond with an error if request body binding fails. apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap()) return } + // Check if the user has the app manager role. if !authverify.IsAppManagerUid(c) { + // Respond with a permission error if the user is not an app manager. apiresp.GinError(c, errs.ErrNoPermission.Wrap("only app manager can send message")) return } + + // Prepare the message request with additional required data. sendMsgReq, err := m.getSendMsgReq(c, req.SendMsg) if err != nil { + // Log and respond with an error if preparation fails. log.ZError(c, "decodeData failed", err) apiresp.GinError(c, err) return } + + // Set the receiver ID in the message data. sendMsgReq.MsgData.RecvID = req.RecvID + + // Declare a variable to store the message sending status. var status int + + // Attempt to send the message using the client. respPb, err := m.Client.SendMsg(c, sendMsgReq) if err != nil { + // Set the status to failed and respond with an error if sending fails. status = constant.MsgSendFailed log.ZError(c, "send message err", err) apiresp.GinError(c, err) return } + + // Set the status to successful if the message is sent. status = constant.MsgSendSuccessed + + // Attempt to update the message sending status in the system. _, err = m.Client.SetSendMsgStatus(c, &msg.SetSendMsgStatusReq{ Status: int32(status), }) if err != nil { + // Log the error if updating the status fails. log.ZError(c, "SetSendMsgStatus failed", err) } + + // Respond with a success message and the response payload. apiresp.GinSuccess(c, respPb) } diff --git a/internal/api/route.go b/internal/api/route.go index fb15a997d..1c91f4dde 100644 --- a/internal/api/route.go +++ b/internal/api/route.go @@ -67,6 +67,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive { userRouterGroup.POST("/user_register", u.UserRegister) userRouterGroup.POST("/update_user_info", ParseToken, u.UpdateUserInfo) + userRouterGroup.POST("/update_user_info_ex", ParseToken, u.UpdateUserInfoEx) userRouterGroup.POST("/set_global_msg_recv_opt", ParseToken, u.SetGlobalRecvMessageOpt) userRouterGroup.POST("/get_users_info", ParseToken, u.GetUsersPublicInfo) userRouterGroup.POST("/get_all_users_uid", ParseToken, u.GetAllUsersID) @@ -107,7 +108,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive friendRouterGroup.POST("/is_friend", f.IsFriend) friendRouterGroup.POST("/get_friend_id", f.GetFriendIDs) friendRouterGroup.POST("/get_specified_friends_info", f.GetSpecifiedFriendsInfo) - //friendRouterGroup.POST("/set_pin_friend", f.SetPinFriends) + friendRouterGroup.POST("/update_friends", f.UpdateFriends) } g := NewGroupApi(*groupRpc) groupRouterGroup := r.Group("/group", ParseToken) diff --git a/internal/api/third.go b/internal/api/third.go index 37ec55098..0a1ef0fbe 100644 --- a/internal/api/third.go +++ b/internal/api/third.go @@ -130,5 +130,5 @@ func (o *ThirdApi) SearchLogs(c *gin.Context) { } func GetPrometheus(c *gin.Context) { - c.Redirect(http.StatusFound, config2.Config.Prometheus.PrometheusUrl) + c.Redirect(http.StatusFound, config2.Config.Prometheus.GrafanaUrl) } diff --git a/internal/api/user.go b/internal/api/user.go index cf68fb422..5f0e23631 100644 --- a/internal/api/user.go +++ b/internal/api/user.go @@ -41,7 +41,9 @@ func (u *UserApi) UserRegister(c *gin.Context) { func (u *UserApi) UpdateUserInfo(c *gin.Context) { a2r.Call(user.UserClient.UpdateUserInfo, u.Client, c) } - +func (u *UserApi) UpdateUserInfoEx(c *gin.Context) { + a2r.Call(user.UserClient.UpdateUserInfoEx, u.Client, c) +} func (u *UserApi) SetGlobalRecvMessageOpt(c *gin.Context) { a2r.Call(user.UserClient.SetGlobalRecvMessageOpt, u.Client, c) } diff --git a/internal/msggateway/client.go b/internal/msggateway/client.go index 69b49d81a..43047fd73 100644 --- a/internal/msggateway/client.go +++ b/internal/msggateway/client.go @@ -87,6 +87,7 @@ func newClient(ctx *UserConnContext, conn LongConn, isCompress bool) *Client { } } +// ResetClient updates the client's state with new connection and context information. func (c *Client) ResetClient( ctx *UserConnContext, conn LongConn, @@ -108,11 +109,13 @@ func (c *Client) ResetClient( c.token = token } +// pingHandler handles ping messages and sends pong responses. func (c *Client) pingHandler(_ string) error { _ = c.conn.SetReadDeadline(pongWait) return c.writePongMsg() } +// readMessage continuously reads messages from the connection. func (c *Client) readMessage() { defer func() { if r := recover(); r != nil { @@ -164,6 +167,7 @@ func (c *Client) readMessage() { } } +// handleMessage processes a single message received by the client. func (c *Client) handleMessage(message []byte) error { if c.IsCompress { var err error diff --git a/internal/msggateway/constant.go b/internal/msggateway/constant.go index fe5f09bdc..045629b4e 100644 --- a/internal/msggateway/constant.go +++ b/internal/msggateway/constant.go @@ -26,6 +26,7 @@ const ( Compression = "compression" GzipCompressionProtocol = "gzip" BackgroundStatus = "isBackground" + MsgResp = "isMsgResp" ) const ( diff --git a/internal/msggateway/n_ws_server.go b/internal/msggateway/n_ws_server.go index 70c2f8fe0..7e8129105 100644 --- a/internal/msggateway/n_ws_server.go +++ b/internal/msggateway/n_ws_server.go @@ -16,7 +16,10 @@ package msggateway import ( "context" + "encoding/json" "errors" + "fmt" + "github.com/OpenIMSDK/tools/apiresp" "net/http" "os" "os/signal" @@ -342,11 +345,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien if !clientOK { return } - - isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients) - if isDeleteUser { - ws.onlineUserNum.Add(-1) - } + ws.clients.deleteClients(newClient.UserID, oldClients) for _, c := range oldClients { err := c.KickOnlineMessage() if err != nil { @@ -422,84 +421,102 @@ func (ws *WsServer) unregisterClient(client *Client) { ) } -func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { - connContext := newContext(w, r) +func (ws *WsServer) ParseWSArgs(r *http.Request) (args *WSArgs, err error) { + var v WSArgs + defer func() { + args = &v + }() + query := r.URL.Query() + v.MsgResp, _ = strconv.ParseBool(query.Get(MsgResp)) if ws.onlineUserConnNum.Load() >= ws.wsMaxConnNum { - httpError(connContext, errs.ErrConnOverMaxNumLimit) - return + return nil, errs.ErrConnOverMaxNumLimit.Wrap("over max conn num limit") } - var ( - token string - userID string - platformIDStr string - exists bool - compression bool - ) - - token, exists = connContext.Query(Token) - if !exists { - httpError(connContext, errs.ErrConnArgsErr) - return + if v.Token = query.Get(Token); v.Token == "" { + return nil, errs.ErrConnArgsErr.Wrap("token is empty") } - userID, exists = connContext.Query(WsUserID) - if !exists { - httpError(connContext, errs.ErrConnArgsErr) - return + if v.UserID = query.Get(WsUserID); v.UserID == "" { + return nil, errs.ErrConnArgsErr.Wrap("sendID is empty") } - platformIDStr, exists = connContext.Query(PlatformID) - if !exists { - httpError(connContext, errs.ErrConnArgsErr) - return + platformIDStr := query.Get(PlatformID) + if platformIDStr == "" { + return nil, errs.ErrConnArgsErr.Wrap("platformID is empty") } platformID, err := strconv.Atoi(platformIDStr) if err != nil { - httpError(connContext, errs.ErrConnArgsErr) - return + return nil, errs.ErrConnArgsErr.Wrap("platformID is not int") } - if err = authverify.WsVerifyToken(token, userID, platformID); err != nil { - httpError(connContext, err) - return + v.PlatformID = platformID + if err = authverify.WsVerifyToken(v.Token, v.UserID, platformID); err != nil { + return nil, err } - m, err := ws.cache.GetTokensWithoutError(context.Background(), userID, platformID) + if query.Get(Compression) == GzipCompressionProtocol { + v.Compression = true + } + if r.Header.Get(Compression) == GzipCompressionProtocol { + v.Compression = true + } + m, err := ws.cache.GetTokensWithoutError(context.Background(), v.UserID, platformID) if err != nil { - httpError(connContext, err) - return + return nil, err } - if v, ok := m[token]; ok { + if v, ok := m[v.Token]; ok { switch v { case constant.NormalToken: case constant.KickedToken: - httpError(connContext, errs.ErrTokenKicked.Wrap()) - return + return nil, errs.ErrTokenKicked.Wrap() default: - httpError(connContext, errs.ErrTokenUnknown.Wrap()) - return + return nil, errs.ErrTokenUnknown.Wrap(fmt.Sprintf("token status is %d", v)) } } else { - httpError(connContext, errs.ErrTokenNotExist.Wrap()) - return + return nil, errs.ErrTokenNotExist.Wrap() } + return &v, nil +} - wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize) - err = wsLongConn.GenerateLongConn(w, r) - if err != nil { - httpError(connContext, err) - return - } - compressProtoc, exists := connContext.Query(Compression) - if exists { - if compressProtoc == GzipCompressionProtocol { - compression = true +type WSArgs struct { + Token string + UserID string + PlatformID int + Compression bool + MsgResp bool +} + +func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { + connContext := newContext(w, r) + args, pErr := ws.ParseWSArgs(r) + var wsLongConn *GWebSocket + if args.MsgResp { + wsLongConn = newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize) + if err := wsLongConn.GenerateLongConn(w, r); err != nil { + httpError(connContext, err) + return } - } - compressProtoc, exists = connContext.GetHeader(Compression) - if exists { - if compressProtoc == GzipCompressionProtocol { - compression = true + data, err := json.Marshal(apiresp.ParseError(pErr)) + if err != nil { + _ = wsLongConn.Close() + return + } + if err := wsLongConn.WriteMessage(MessageText, data); err != nil { + _ = wsLongConn.Close() + return + } + if pErr != nil { + _ = wsLongConn.Close() + return + } + } else { + if pErr != nil { + httpError(connContext, pErr) + return + } + wsLongConn = newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize) + if err := wsLongConn.GenerateLongConn(w, r); err != nil { + httpError(connContext, err) + return } } client := ws.clientPool.Get().(*Client) - client.ResetClient(connContext, wsLongConn, connContext.GetBackground(), compression, ws, token) + client.ResetClient(connContext, wsLongConn, connContext.GetBackground(), args.Compression, ws, args.Token) ws.registerChan <- client go client.readMessage() } diff --git a/internal/rpc/friend/friend.go b/internal/rpc/friend/friend.go index c40b566f3..c53cb88f5 100644 --- a/internal/rpc/friend/friend.go +++ b/internal/rpc/friend/friend.go @@ -53,10 +53,6 @@ type friendServer struct { RegisterCenter registry.SvcDiscoveryRegistry } -func (s *friendServer) UpdateFriends(ctx context.Context, req *pbfriend.UpdateFriendsReq) (*pbfriend.UpdateFriendsResp, error) { - return nil, errs.ErrInternalServer.Wrap("not implemented") -} - func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error { // Initialize MongoDB mongo, err := unrelation.NewMongo() @@ -440,7 +436,7 @@ func (s *friendServer) GetSpecifiedFriendsInfo(ctx context.Context, req *pbfrien } return resp, nil } -func (s *friendServer) PinFriends( +func (s *friendServer) UpdateFriends( ctx context.Context, req *pbfriend.UpdateFriendsReq, ) (*pbfriend.UpdateFriendsResp, error) { @@ -450,25 +446,35 @@ func (s *friendServer) PinFriends( if utils.Duplicate(req.FriendUserIDs) { return nil, errs.ErrArgs.Wrap("friendIDList repeated") } - var isPinned bool - if req.IsPinned != nil { - isPinned = req.IsPinned.Value - } else { - return nil, errs.ErrArgs.Wrap("isPinned is nil") - } - //check whther in friend list + _, err := s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs) if err != nil { return nil, err } - //set friendslist friend pin status to isPinned for _, friendID := range req.FriendUserIDs { - if err := s.friendDatabase.UpdateFriendPinStatus(ctx, req.OwnerUserID, friendID, isPinned); err != nil { - return nil, err + if req.IsPinned != nil { + if err = s.friendDatabase.UpdateFriendPinStatus(ctx, req.OwnerUserID, friendID, req.IsPinned.Value); err != nil { + return nil, err + } + } + if req.Remark != nil { + if err = s.friendDatabase.UpdateFriendRemark(ctx, req.OwnerUserID, friendID, req.Remark.Value); err != nil { + return nil, err + } + } + if req.Ex != nil { + if err = s.friendDatabase.UpdateFriendEx(ctx, req.OwnerUserID, friendID, req.Ex.Value); err != nil { + return nil, err + } } } resp := &pbfriend.UpdateFriendsResp{} + + err = s.notificationSender.FriendsInfoUpdateNotification(ctx, req.OwnerUserID, req.FriendUserIDs) + if err != nil { + return nil, errs.Wrap(err, "FriendsInfoUpdateNotification Error") + } return resp, nil } diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index 93485ff75..4af4d3c03 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -1051,20 +1051,28 @@ func (s *groupServer) TransferGroupOwner(ctx context.Context, req *pbgroup.Trans func (s *groupServer) GetGroups(ctx context.Context, req *pbgroup.GetGroupsReq) (*pbgroup.GetGroupsResp, error) { resp := &pbgroup.GetGroupsResp{} var ( - groups []*relationtb.GroupModel - err error + group []*relationtb.GroupModel + err error ) if req.GroupID != "" { - groups, err = s.db.FindGroup(ctx, []string{req.GroupID}) - resp.Total = uint32(len(groups)) + group, err = s.db.FindGroup(ctx, []string{req.GroupID}) + resp.Total = uint32(len(group)) } else { var total int64 - total, groups, err = s.db.SearchGroup(ctx, req.GroupName, req.Pagination) + total, group, err = s.db.SearchGroup(ctx, req.GroupName, req.Pagination) resp.Total = uint32(total) } if err != nil { return nil, err } + + var groups []*relationtb.GroupModel + for _, v := range group { + if v.Status == constant.GroupStatusDismissed { + continue + } + groups = append(groups, v) + } groupIDs := utils.Slice(groups, func(e *relationtb.GroupModel) string { return e.GroupID }) diff --git a/internal/rpc/msg/callback.go b/internal/rpc/msg/callback.go index 3a8587ea3..5d192fb87 100644 --- a/internal/rpc/msg/callback.go +++ b/internal/rpc/msg/callback.go @@ -100,7 +100,7 @@ func callbackAfterSendSingleMsg(ctx context.Context, msg *pbchat.SendMsgReq) err } func callbackBeforeSendGroupMsg(ctx context.Context, msg *pbchat.SendMsgReq) error { - if !config.Config.Callback.CallbackBeforeSendSingleMsg.Enable { + if !config.Config.Callback.CallbackBeforeSendGroupMsg.Enable { return nil } req := &cbapi.CallbackBeforeSendGroupMsgReq{ diff --git a/internal/rpc/user/callback.go b/internal/rpc/user/callback.go index 7d865af5f..092a66a07 100644 --- a/internal/rpc/user/callback.go +++ b/internal/rpc/user/callback.go @@ -16,7 +16,6 @@ package user import ( "context" - pbuser "github.com/OpenIMSDK/protocol/user" "github.com/OpenIMSDK/tools/utils" @@ -44,7 +43,6 @@ func CallbackBeforeUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInf utils.NotNilReplace(&req.UserInfo.Nickname, resp.Nickname) return nil } - func CallbackAfterUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfoReq) error { if !config.Config.Callback.CallbackAfterUpdateUserInfo.Enable { return nil @@ -61,6 +59,41 @@ func CallbackAfterUpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfo } return nil } +func CallbackBeforeUpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) error { + if !config.Config.Callback.CallbackBeforeUpdateUserInfoEx.Enable { + return nil + } + cbReq := &cbapi.CallbackBeforeUpdateUserInfoExReq{ + CallbackCommand: cbapi.CallbackBeforeUpdateUserInfoExCommand, + UserID: req.UserInfo.UserID, + FaceURL: req.UserInfo.FaceURL, + Nickname: req.UserInfo.Nickname, + } + resp := &cbapi.CallbackBeforeUpdateUserInfoExResp{} + if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfoEx); err != nil { + return err + } + utils.NotNilReplace(req.UserInfo.FaceURL, resp.FaceURL) + utils.NotNilReplace(req.UserInfo.Ex, resp.Ex) + utils.NotNilReplace(req.UserInfo.Nickname, resp.Nickname) + return nil +} +func CallbackAfterUpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) error { + if !config.Config.Callback.CallbackAfterUpdateUserInfoEx.Enable { + return nil + } + cbReq := &cbapi.CallbackAfterUpdateUserInfoExReq{ + CallbackCommand: cbapi.CallbackAfterUpdateUserInfoExCommand, + UserID: req.UserInfo.UserID, + FaceURL: req.UserInfo.FaceURL, + Nickname: req.UserInfo.Nickname, + } + resp := &cbapi.CallbackAfterUpdateUserInfoExResp{} + if err := http.CallBackPostReturn(ctx, config.Config.Callback.CallbackUrl, cbReq, resp, config.Config.Callback.CallbackBeforeUpdateUserInfoEx); err != nil { + return err + } + return nil +} func CallbackBeforeUserRegister(ctx context.Context, req *pbuser.UserRegisterReq) error { if !config.Config.Callback.CallbackBeforeUserRegister.Enable { diff --git a/internal/rpc/user/user.go b/internal/rpc/user/user.go index b5ad186a5..51403d631 100644 --- a/internal/rpc/user/user.go +++ b/internal/rpc/user/user.go @@ -17,6 +17,7 @@ package user import ( "context" "errors" + "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation" "math/rand" "strings" "time" @@ -57,10 +58,6 @@ type userServer struct { RegisterCenter registry.SvcDiscoveryRegistry } -func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) (*pbuser.UpdateUserInfoExResp, error) { - return nil, errs.ErrInternalServer.Wrap("not implemented") -} - func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error { rdb, err := cache.NewRedis() if err != nil { @@ -152,7 +149,41 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserI } return resp, nil } +func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUserInfoExReq) (resp *pbuser.UpdateUserInfoExResp, err error) { + resp = &pbuser.UpdateUserInfoExResp{} + err = authverify.CheckAccessV3(ctx, req.UserInfo.UserID) + if err != nil { + return nil, err + } + if err = CallbackBeforeUpdateUserInfoEx(ctx, req); err != nil { + return nil, err + } + data := convert.UserPb2DBMapEx(req.UserInfo) + if err = s.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil { + return nil, err + } + _ = s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID) + friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID) + if err != nil { + return nil, err + } + if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil { + if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { + log.ZError(ctx, "NotificationUserInfoUpdate", err) + } + } + for _, friendID := range friends { + s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) + } + if err := CallbackAfterUpdateUserInfoEx(ctx, req); err != nil { + return nil, err + } + if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { + log.ZError(ctx, "NotificationUserInfoUpdate", err, "userID", req.UserInfo.UserID) + } + return resp, nil +} func (s *userServer) SetGlobalRecvMessageOpt(ctx context.Context, req *pbuser.SetGlobalRecvMessageOptReq) (resp *pbuser.SetGlobalRecvMessageOptResp, err error) { resp = &pbuser.SetGlobalRecvMessageOptResp{} if _, err := s.FindWithError(ctx, []string{req.UserID}); err != nil { @@ -466,26 +497,31 @@ func (s *userServer) SearchNotificationAccount(ctx context.Context, req *pbuser. return nil, err } - _, users, err := s.UserDatabase.Page(ctx, req.Pagination) - if err != nil { - return nil, err + if req.NickName != "" { + users, err := s.UserDatabase.FindByNickname(ctx, req.NickName) + if err != nil { + return nil, err + } + resp := s.userModelToResp(users) + return resp, nil } - var total int64 - accounts := make([]*pbuser.NotificationAccountInfo, 0, len(users)) - for _, v := range users { - if v.AppMangerLevel != constant.AppNotificationAdmin { - continue - } - temp := &pbuser.NotificationAccountInfo{ - UserID: v.UserID, - FaceURL: v.FaceURL, - NickName: v.Nickname, + if req.UserID != "" { + users, err := s.UserDatabase.Find(ctx, []string{req.UserID}) + if err != nil { + return nil, err } - accounts = append(accounts, temp) - total += 1 + resp := s.userModelToResp(users) + return resp, nil } - return &pbuser.SearchNotificationAccountResp{Total: total, NotificationAccounts: accounts}, nil + + users, err := s.UserDatabase.FindNotification(ctx, constant.AppNotificationAdmin) + if err != nil { + return nil, err + } + + resp := s.userModelToResp(users) + return resp, nil } func (s *userServer) GetNotificationAccount(ctx context.Context, req *pbuser.GetNotificationAccountReq) (*pbuser.GetNotificationAccountResp, error) { @@ -517,3 +553,20 @@ func (s *userServer) genUserID() string { } return string(data) } + +func (s *userServer) userModelToResp(users []*relation.UserModel) *pbuser.SearchNotificationAccountResp { + accounts := make([]*pbuser.NotificationAccountInfo, 0) + var total int64 + for _, v := range users { + if v.AppMangerLevel == constant.AppNotificationAdmin && !utils.IsContain(v.UserID, config.Config.IMAdmin.UserID) { + temp := &pbuser.NotificationAccountInfo{ + UserID: v.UserID, + FaceURL: v.FaceURL, + NickName: v.Nickname, + } + accounts = append(accounts, temp) + total += 1 + } + } + return &pbuser.SearchNotificationAccountResp{Total: total, NotificationAccounts: accounts} +} diff --git a/pkg/authverify/token.go b/pkg/authverify/token.go index d9aa0dbb1..e810dfecf 100644 --- a/pkg/authverify/token.go +++ b/pkg/authverify/token.go @@ -38,6 +38,9 @@ func CheckAccessV3(ctx context.Context, ownerUserID string) (err error) { if utils.IsContain(opUserID, config.Config.Manager.UserID) { return nil } + if utils.IsContain(opUserID, config.Config.IMAdmin.UserID) { + return nil + } if opUserID == ownerUserID { return nil } @@ -45,13 +48,16 @@ func CheckAccessV3(ctx context.Context, ownerUserID string) (err error) { } func IsAppManagerUid(ctx context.Context) bool { - return utils.IsContain(mcontext.GetOpUserID(ctx), config.Config.Manager.UserID) + return utils.IsContain(mcontext.GetOpUserID(ctx), config.Config.Manager.UserID) || utils.IsContain(mcontext.GetOpUserID(ctx), config.Config.IMAdmin.UserID) } func CheckAdmin(ctx context.Context) error { if utils.IsContain(mcontext.GetOpUserID(ctx), config.Config.Manager.UserID) { return nil } + if utils.IsContain(mcontext.GetOpUserID(ctx), config.Config.IMAdmin.UserID) { + return nil + } return errs.ErrNoPermission.Wrap(fmt.Sprintf("user %s is not admin userID", mcontext.GetOpUserID(ctx))) } func CheckIMAdmin(ctx context.Context) error { @@ -69,7 +75,8 @@ func ParseRedisInterfaceToken(redisToken any) (*tokenverify.Claims, error) { } func IsManagerUserID(opUserID string) bool { - return utils.IsContain(opUserID, config.Config.Manager.UserID) + return utils.IsContain(opUserID, config.Config.Manager.UserID) || utils.IsContain(opUserID, config.Config.IMAdmin.UserID) + } func WsVerifyToken(token, userID string, platformID int) error { diff --git a/pkg/callbackstruct/constant.go b/pkg/callbackstruct/constant.go index 0a03a1ef1..cda98af16 100644 --- a/pkg/callbackstruct/constant.go +++ b/pkg/callbackstruct/constant.go @@ -37,6 +37,8 @@ const ( CallbackGroupMsgReadCommand = "callbackGroupMsgReadCommand" CallbackMsgModifyCommand = "callbackMsgModifyCommand" CallbackAfterUpdateUserInfoCommand = "callbackAfterUpdateUserInfoCommand" + CallbackAfterUpdateUserInfoExCommand = "callbackAfterUpdateUserInfoExCommand" + CallbackBeforeUpdateUserInfoExCommand = "callbackBeforeUpdateUserInfoExCommand" CallbackBeforeUserRegisterCommand = "callbackBeforeUserRegisterCommand" CallbackAfterUserRegisterCommand = "callbackAfterUserRegisterCommand" CallbackTransferGroupOwnerAfter = "callbackTransferGroupOwnerAfter" diff --git a/pkg/callbackstruct/user.go b/pkg/callbackstruct/user.go index f35cff554..98536882d 100644 --- a/pkg/callbackstruct/user.go +++ b/pkg/callbackstruct/user.go @@ -14,7 +14,10 @@ package callbackstruct -import "github.com/OpenIMSDK/protocol/sdkws" +import ( + "github.com/OpenIMSDK/protocol/sdkws" + "github.com/OpenIMSDK/protocol/wrapperspb" +) type CallbackBeforeUpdateUserInfoReq struct { CallbackCommand `json:"callbackCommand"` @@ -41,6 +44,31 @@ type CallbackAfterUpdateUserInfoResp struct { CommonCallbackResp } +type CallbackBeforeUpdateUserInfoExReq struct { + CallbackCommand `json:"callbackCommand"` + UserID string `json:"userID"` + Nickname *wrapperspb.StringValue `json:"nickName"` + FaceURL *wrapperspb.StringValue `json:"faceURL"` + Ex *wrapperspb.StringValue `json:"ex"` +} +type CallbackBeforeUpdateUserInfoExResp struct { + CommonCallbackResp + Nickname *wrapperspb.StringValue `json:"nickName"` + FaceURL *wrapperspb.StringValue `json:"faceURL"` + Ex *wrapperspb.StringValue `json:"ex"` +} + +type CallbackAfterUpdateUserInfoExReq struct { + CallbackCommand `json:"callbackCommand"` + UserID string `json:"userID"` + Nickname *wrapperspb.StringValue `json:"nickName"` + FaceURL *wrapperspb.StringValue `json:"faceURL"` + Ex *wrapperspb.StringValue `json:"ex"` +} +type CallbackAfterUpdateUserInfoExResp struct { + CommonCallbackResp +} + type CallbackBeforeUserRegisterReq struct { CallbackCommand `json:"callbackCommand"` Secret string `json:"secret"` diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index 707601234..88e87e709 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -282,6 +282,8 @@ type configStruct struct { CallbackBeforeSetFriendRemark CallBackConfig `yaml:"callbackBeforeSetFriendRemark"` CallbackAfterSetFriendRemark CallBackConfig `yaml:"callbackAfterSetFriendRemark"` CallbackBeforeUpdateUserInfo CallBackConfig `yaml:"beforeUpdateUserInfo"` + CallbackBeforeUpdateUserInfoEx CallBackConfig `yaml:"beforeUpdateUserInfoEx"` + CallbackAfterUpdateUserInfoEx CallBackConfig `yaml:"afterUpdateUserInfoEx"` CallbackBeforeUserRegister CallBackConfig `yaml:"beforeUserRegister"` CallbackAfterUpdateUserInfo CallBackConfig `yaml:"updateUserInfo"` CallbackAfterUserRegister CallBackConfig `yaml:"afterUserRegister"` @@ -312,7 +314,7 @@ type configStruct struct { Prometheus struct { Enable bool `yaml:"enable"` - PrometheusUrl string `yaml:"prometheusUrl"` + GrafanaUrl string `yaml:"grafanaUrl"` ApiPrometheusPort []int `yaml:"apiPrometheusPort"` UserPrometheusPort []int `yaml:"userPrometheusPort"` FriendPrometheusPort []int `yaml:"friendPrometheusPort"` diff --git a/pkg/common/convert/user.go b/pkg/common/convert/user.go index 72041a790..62f80e458 100644 --- a/pkg/common/convert/user.go +++ b/pkg/common/convert/user.go @@ -64,7 +64,7 @@ func UserPb2DBMap(user *sdkws.UserInfo) map[string]any { "global_recv_msg_opt": user.GlobalRecvMsgOpt, } for key, value := range fields { - if v, ok := value.(string); ok { + if v, ok := value.(string); ok && v != "" { val[key] = v } else if v, ok := value.(int32); ok && v != 0 { val[key] = v @@ -72,3 +72,25 @@ func UserPb2DBMap(user *sdkws.UserInfo) map[string]any { } return val } +func UserPb2DBMapEx(user *sdkws.UserInfoWithEx) map[string]any { + if user == nil { + return nil + } + val := make(map[string]any) + + // Map fields from UserInfoWithEx to val + if user.Nickname != nil { + val["nickname"] = user.Nickname.Value + } + if user.FaceURL != nil { + val["face_url"] = user.FaceURL.Value + } + if user.Ex != nil { + val["ex"] = user.Ex.Value + } + if user.GlobalRecvMsgOpt != nil { + val["global_recv_msg_opt"] = user.GlobalRecvMsgOpt.Value + } + + return val +} diff --git a/pkg/common/db/cache/init_redis.go b/pkg/common/db/cache/init_redis.go index 63d2f5707..1308e9649 100644 --- a/pkg/common/db/cache/init_redis.go +++ b/pkg/common/db/cache/init_redis.go @@ -87,7 +87,15 @@ func NewRedis() (redis.UniversalClient, error) { // overrideConfigFromEnv overrides configuration fields with environment variables if present. func overrideConfigFromEnv() { if envAddr := os.Getenv("REDIS_ADDRESS"); envAddr != "" { - config.Config.Redis.Address = strings.Split(envAddr, ",") // Assuming addresses are comma-separated + if envPort := os.Getenv("REDIS_PORT"); envPort != "" { + addresses := strings.Split(envAddr, ",") + for i, addr := range addresses { + addresses[i] = addr + ":" + envPort + } + config.Config.Redis.Address = addresses + } else { + config.Config.Redis.Address = strings.Split(envAddr, ",") + } } if envUser := os.Getenv("REDIS_USERNAME"); envUser != "" { config.Config.Redis.Username = envUser diff --git a/pkg/common/db/controller/friend.go b/pkg/common/db/controller/friend.go index 34ce22295..924a179ba 100644 --- a/pkg/common/db/controller/friend.go +++ b/pkg/common/db/controller/friend.go @@ -32,33 +32,57 @@ import ( ) type FriendDatabase interface { - // 检查user2是否在user1的好友列表中(inUser1Friends==true) 检查user1是否在user2的好友列表中(inUser2Friends==true) + // CheckIn checks if user2 is in user1's friend list (inUser1Friends==true) and if user1 is in user2's friend list (inUser2Friends==true) CheckIn(ctx context.Context, user1, user2 string) (inUser1Friends bool, inUser2Friends bool, err error) - // 增加或者更新好友申请 + + // AddFriendRequest adds or updates a friend request AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error) - // 先判断是否在好友表,如果在则不插入 + + // BecomeFriends first checks if the users are already in the friends table; if not, it inserts them as friends BecomeFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, addSource int32) (err error) - // 拒绝好友申请 + + // RefuseFriendRequest refuses a friend request RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) - // 同意好友申请 + + // AgreeFriendRequest accepts a friend request AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) - // 删除好友 + + // Delete removes a friend or friends from the owner's friend list Delete(ctx context.Context, ownerUserID string, friendUserIDs []string) (err error) - // 更新好友备注 + + // UpdateRemark updates the remark for a friend UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) - // 获取ownerUserID的好友列表 + + // PageOwnerFriends retrieves the friend list of ownerUserID with pagination PageOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendModel, err error) - // friendUserID在哪些人的好友列表中 + + // PageInWhoseFriends finds the users who have friendUserID in their friend list with pagination PageInWhoseFriends(ctx context.Context, friendUserID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendModel, err error) - // 获取我发出去的好友申请 + + // PageFriendRequestFromMe retrieves the friend requests sent by the user with pagination PageFriendRequestFromMe(ctx context.Context, userID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendRequestModel, err error) - // 获取我收到的的好友申请 + + // PageFriendRequestToMe retrieves the friend requests received by the user with pagination PageFriendRequestToMe(ctx context.Context, userID string, pagination pagination.Pagination) (total int64, friends []*relation.FriendRequestModel, err error) - // 获取某人指定好友的信息 + + // FindFriendsWithError fetches specified friends of a user and returns an error if any do not exist FindFriendsWithError(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error) + + // FindFriendUserIDs retrieves the friend IDs of a user FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error) + + // FindBothFriendRequests finds friend requests sent and received FindBothFriendRequests(ctx context.Context, fromUserID, toUserID string) (friends []*relation.FriendRequestModel, err error) + + // UpdateFriendPinStatus updates the pinned status of a friend UpdateFriendPinStatus(ctx context.Context, ownerUserID string, friendUserID string, isPinned bool) (err error) + + // UpdateFriendRemark updates the remark for a friend + UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error) + + // UpdateFriendEx updates the 'ex' field for a friend + UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error) + } type friendDatabase struct { @@ -305,3 +329,15 @@ func (f *friendDatabase) UpdateFriendPinStatus(ctx context.Context, ownerUserID } return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx) } +func (f *friendDatabase) UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error) { + if err := f.friend.UpdateFriendRemark(ctx, ownerUserID, friendUserID, remark); err != nil { + return err + } + return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx) +} +func (f *friendDatabase) UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error) { + if err := f.friend.UpdateFriendEx(ctx, ownerUserID, friendUserID, ex); err != nil { + return err + } + return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx) +} diff --git a/pkg/common/db/controller/group.go b/pkg/common/db/controller/group.go index b7ff30401..4fe586b0d 100644 --- a/pkg/common/db/controller/group.go +++ b/pkg/common/db/controller/group.go @@ -226,7 +226,7 @@ func (g *groupDatabase) UpdateGroup(ctx context.Context, groupID string, data ma func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, deleteMember bool) error { return g.ctxTx.Transaction(ctx, func(ctx context.Context) error { c := g.cache.NewCache() - if err := g.groupDB.UpdateState(ctx, groupID, constant.GroupStatusDismissed); err != nil { + if err := g.groupDB.UpdateStatus(ctx, groupID, constant.GroupStatusDismissed); err != nil { return err } if deleteMember { diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index a7230db00..dfff5c61d 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -18,7 +18,6 @@ import ( "context" "encoding/json" "errors" - "fmt" "time" "github.com/OpenIMSDK/protocol/constant" @@ -1056,15 +1055,13 @@ func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbmsg.Searc func (db *commonMsgDatabase) FindOneByDocIDs(ctx context.Context, conversationIDs []string, seqs map[string]int64) (map[string]*sdkws.MsgData, error) { totalMsgs := make(map[string]*sdkws.MsgData) for _, conversationID := range conversationIDs { - len := seqs[conversationID] - seq := len / 100 - index := (len % 100) - 1 - - docID := conversationID + ":" + fmt.Sprintf("%d", seq) + seq := seqs[conversationID] + docID := db.msg.GetDocID(conversationID, seq) msgs, err := db.msgDocDatabase.FindOneByDocID(ctx, docID) if err != nil { return nil, err } + index := db.msg.GetMsgIndex(seq) totalMsgs[conversationID] = convert.MsgDB2Pb(msgs.Msg[index].Msg) } return totalMsgs, nil diff --git a/pkg/common/db/controller/user.go b/pkg/common/db/controller/user.go index 2fafcf266..a109b81ef 100644 --- a/pkg/common/db/controller/user.go +++ b/pkg/common/db/controller/user.go @@ -38,6 +38,10 @@ type UserDatabase interface { FindWithError(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error) // Find Get the information of the specified user If the userID is not found, no error will be returned Find(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error) + // Find userInfo By Nickname + FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error) + // Find notificationAccounts + FindNotification(ctx context.Context, level int64) (users []*relation.UserModel, err error) // Create Insert multiple external guarantees that the userID is not repeated and does not exist in the db Create(ctx context.Context, users []*relation.UserModel) (err error) // Update update (non-zero value) external guarantee userID exists @@ -133,6 +137,16 @@ func (u *userDatabase) Find(ctx context.Context, userIDs []string) (users []*rel return u.cache.GetUsersInfo(ctx, userIDs) } +// Find userInfo By Nickname +func (u *userDatabase) FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error) { + return u.userDB.TakeByNickname(ctx, nickname) +} + +// Find notificationAccouts +func (u *userDatabase) FindNotification(ctx context.Context, level int64) (users []*relation.UserModel, err error) { + return u.userDB.TakeNotification(ctx, level) +} + // Create Insert multiple external guarantees that the userID is not repeated and does not exist in the db. func (u *userDatabase) Create(ctx context.Context, users []*relation.UserModel) (err error) { return u.tx.Transaction(ctx, func(ctx context.Context) error { diff --git a/pkg/common/db/mgo/friend.go b/pkg/common/db/mgo/friend.go index 667098819..72289181b 100644 --- a/pkg/common/db/mgo/friend.go +++ b/pkg/common/db/mgo/friend.go @@ -160,3 +160,33 @@ func (f *FriendMgo) UpdatePinStatus(ctx context.Context, ownerUserID string, fri return nil } +func (f *FriendMgo) UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error) { + + filter := bson.M{"owner_user_id": ownerUserID, "friend_user_id": friendUserID} + // Create an update operation to set the "is_pinned" field to isPinned for all documents. + update := bson.M{"$set": bson.M{"remark": remark}} + + // Perform the update operation for all documents in the collection. + _, err = f.coll.UpdateMany(ctx, filter, update) + + if err != nil { + return errs.Wrap(err, "update remark error") + } + + return nil +} +func (f *FriendMgo) UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error) { + + filter := bson.M{"owner_user_id": ownerUserID, "friend_user_id": friendUserID} + // Create an update operation to set the "is_pinned" field to isPinned for all documents. + update := bson.M{"$set": bson.M{"ex": ex}} + + // Perform the update operation for all documents in the collection. + _, err = f.coll.UpdateMany(ctx, filter, update) + + if err != nil { + return errs.Wrap(err, "update ex error") + } + + return nil +} diff --git a/pkg/common/db/mgo/group.go b/pkg/common/db/mgo/group.go index 65dbbca59..a9c6d1eb8 100644 --- a/pkg/common/db/mgo/group.go +++ b/pkg/common/db/mgo/group.go @@ -49,8 +49,8 @@ func (g *GroupMgo) Create(ctx context.Context, groups []*relation.GroupModel) (e return mgoutil.InsertMany(ctx, g.coll, groups) } -func (g *GroupMgo) UpdateState(ctx context.Context, groupID string, state int32) (err error) { - return g.UpdateMap(ctx, groupID, map[string]any{"state": state}) +func (g *GroupMgo) UpdateStatus(ctx context.Context, groupID string, status int32) (err error) { + return g.UpdateMap(ctx, groupID, map[string]any{"status": status}) } func (g *GroupMgo) UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error) { diff --git a/pkg/common/db/mgo/group_member.go b/pkg/common/db/mgo/group_member.go index 8c3041901..8e3dd1efa 100644 --- a/pkg/common/db/mgo/group_member.go +++ b/pkg/common/db/mgo/group_member.go @@ -51,7 +51,11 @@ func (g *GroupMemberMgo) Create(ctx context.Context, groupMembers []*relation.Gr } func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []string) (err error) { - return mgoutil.DeleteMany(ctx, g.coll, bson.M{"group_id": groupID, "user_id": bson.M{"$in": userIDs}}) + filter := bson.M{"group_id": groupID} + if len(userIDs) > 0 { + filter["user_id"] = bson.M{"$in": userIDs} + } + return mgoutil.DeleteMany(ctx, g.coll, filter) } func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, userID string, roleLevel int32) error { @@ -84,8 +88,8 @@ func (g *GroupMemberMgo) FindRoleLevelUserIDs(ctx context.Context, groupID strin } func (g *GroupMemberMgo) SearchMember(ctx context.Context, keyword string, groupID string, pagination pagination.Pagination) (total int64, groupList []*relation.GroupMemberModel, err error) { - //TODO implement me - panic("implement me") + filter := bson.M{"group_id": groupID, "nickname": bson.M{"$regex": keyword}} + return mgoutil.FindPage[*relation.GroupMemberModel](ctx, g.coll, filter, pagination) } func (g *GroupMemberMgo) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) { diff --git a/pkg/common/db/mgo/user.go b/pkg/common/db/mgo/user.go index 268d69ce3..66b962d9d 100644 --- a/pkg/common/db/mgo/user.go +++ b/pkg/common/db/mgo/user.go @@ -17,6 +17,7 @@ package mgo import ( "context" "github.com/OpenIMSDK/protocol/user" + "go.mongodb.org/mongo-driver/bson/primitive" "time" "github.com/OpenIMSDK/tools/mgoutil" @@ -65,6 +66,14 @@ func (u *UserMgo) Take(ctx context.Context, userID string) (user *relation.UserM return mgoutil.FindOne[*relation.UserModel](ctx, u.coll, bson.M{"user_id": userID}) } +func (u *UserMgo) TakeNotification(ctx context.Context, level int64) (user []*relation.UserModel, err error) { + return mgoutil.Find[*relation.UserModel](ctx, u.coll, bson.M{"app_manger_level": level}) +} + +func (u *UserMgo) TakeByNickname(ctx context.Context, nickname string) (user []*relation.UserModel, err error) { + return mgoutil.Find[*relation.UserModel](ctx, u.coll, bson.E{Key: "nickname", Value: bson.M{"$regex": primitive.Regex{Pattern: ".*" + nickname + ".*", Options: "i"}}}) +} + func (u *UserMgo) Page(ctx context.Context, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) { return mgoutil.FindPage[*relation.UserModel](ctx, u.coll, bson.M{}, pagination) } diff --git a/pkg/common/db/table/relation/friend.go b/pkg/common/db/table/relation/friend.go index 4f85998f4..cc337701d 100644 --- a/pkg/common/db/table/relation/friend.go +++ b/pkg/common/db/table/relation/friend.go @@ -59,4 +59,8 @@ type FriendModelInterface interface { FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error) // UpdatePinStatus update friend's pin status UpdatePinStatus(ctx context.Context, ownerUserID string, friendUserID string, isPinned bool) (err error) + // UpdateFriendRemark update friend's remark + UpdateFriendRemark(ctx context.Context, ownerUserID string, friendUserID string, remark string) (err error) + // UpdateFriendEx update friend's ex + UpdateFriendEx(ctx context.Context, ownerUserID string, friendUserID string, ex string) (err error) } diff --git a/pkg/common/db/table/relation/group.go b/pkg/common/db/table/relation/group.go index bb1ddd878..57d6b1d62 100644 --- a/pkg/common/db/table/relation/group.go +++ b/pkg/common/db/table/relation/group.go @@ -42,7 +42,7 @@ type GroupModel struct { type GroupModelInterface interface { Create(ctx context.Context, groups []*GroupModel) (err error) UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error) - UpdateState(ctx context.Context, groupID string, state int32) (err error) + UpdateStatus(ctx context.Context, groupID string, status int32) (err error) Find(ctx context.Context, groupIDs []string) (groups []*GroupModel, err error) Take(ctx context.Context, groupID string) (group *GroupModel, err error) Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, groups []*GroupModel, err error) diff --git a/pkg/common/db/table/relation/user.go b/pkg/common/db/table/relation/user.go index 213a6b774..fc116adc2 100644 --- a/pkg/common/db/table/relation/user.go +++ b/pkg/common/db/table/relation/user.go @@ -53,6 +53,8 @@ type UserModelInterface interface { UpdateByMap(ctx context.Context, userID string, args map[string]any) (err error) Find(ctx context.Context, userIDs []string) (users []*UserModel, err error) Take(ctx context.Context, userID string) (user *UserModel, err error) + TakeNotification(ctx context.Context, level int64) (user []*UserModel, err error) + TakeByNickname(ctx context.Context, nickname string) (user []*UserModel, err error) Page(ctx context.Context, pagination pagination.Pagination) (count int64, users []*UserModel, err error) Exist(ctx context.Context, userID string) (exist bool, err error) GetAllUserID(ctx context.Context, pagination pagination.Pagination) (count int64, userIDs []string, err error) diff --git a/pkg/common/db/unrelation/mongo.go b/pkg/common/db/unrelation/mongo.go index 2c1dc6f38..8cfb97a98 100644 --- a/pkg/common/db/unrelation/mongo.go +++ b/pkg/common/db/unrelation/mongo.go @@ -74,6 +74,10 @@ func buildMongoURI() string { return uri } + if config.Config.Mongo.Uri != "" { + return config.Config.Mongo.Uri + } + username := os.Getenv("MONGO_USERNAME") password := os.Getenv("MONGO_PASSWORD") address := os.Getenv("MONGO_ADDRESS") diff --git a/pkg/common/discoveryregister/discoveryregister_test.go b/pkg/common/discoveryregister/discoveryregister_test.go index d4a634b91..d83da1285 100644 --- a/pkg/common/discoveryregister/discoveryregister_test.go +++ b/pkg/common/discoveryregister/discoveryregister_test.go @@ -24,7 +24,8 @@ import ( func setupTestEnvironment() { os.Setenv("ZOOKEEPER_SCHEMA", "openim") - os.Setenv("ZOOKEEPER_ADDRESS", "172.28.0.1:12181") + os.Setenv("ZOOKEEPER_ADDRESS", "172.28.0.1") + os.Setenv("ZOOKEEPER_PORT", "12181") os.Setenv("ZOOKEEPER_USERNAME", "") os.Setenv("ZOOKEEPER_PASSWORD", "") } diff --git a/pkg/common/discoveryregister/kubernetes/kubernetes.go b/pkg/common/discoveryregister/kubernetes/kubernetes.go index 2ff1539e6..c10518056 100644 --- a/pkg/common/discoveryregister/kubernetes/kubernetes.go +++ b/pkg/common/discoveryregister/kubernetes/kubernetes.go @@ -55,14 +55,17 @@ func (cli *K8sDR) Register(serviceName, host string, port int, opts ...grpc.Dial return nil } + func (cli *K8sDR) UnRegister() error { return nil } + func (cli *K8sDR) CreateRpcRootNodes(serviceNames []string) error { return nil } + func (cli *K8sDR) RegisterConf2Registry(key string, conf []byte) error { return nil @@ -123,6 +126,8 @@ func getMsgGatewayHost(ctx context.Context) []string { log.ZInfo(ctx, "getMsgGatewayHost", "instance", instance, "selfPodName", selfPodName, "replicas", replicas, "ns", ns, "ret", ret) return ret } + +// GetConns returns the gRPC client connections to the specified service. func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) { if serviceName != config.Config.RpcRegisterName.OpenImMessageGatewayName { @@ -142,6 +147,7 @@ func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc return ret, nil } } + func (cli *K8sDR) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { return grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...) @@ -151,9 +157,11 @@ func (cli *K8sDR) GetSelfConnTarget() string { return cli.rpcRegisterAddr } + func (cli *K8sDR) AddOption(opts ...grpc.DialOption) { cli.options = append(cli.options, opts...) } + func (cli *K8sDR) CloseConn(conn *grpc.ClientConn) { conn.Close() } diff --git a/pkg/common/discoveryregister/zookeeper/zookeeper.go b/pkg/common/discoveryregister/zookeeper/zookeeper.go index a4edffc43..db1a5c6c4 100644 --- a/pkg/common/discoveryregister/zookeeper/zookeeper.go +++ b/pkg/common/discoveryregister/zookeeper/zookeeper.go @@ -52,10 +52,18 @@ func getEnv(key, fallback string) string { return fallback } -// getZkAddrFromEnv returns the value of an environment variable if it exists, otherwise it returns the fallback value. +// getZkAddrFromEnv returns the Zookeeper addresses combined from the ZOOKEEPER_ADDRESS and ZOOKEEPER_PORT environment variables. +// If the environment variables are not set, it returns the fallback value. func getZkAddrFromEnv(fallback []string) []string { - if value, exists := os.LookupEnv("ZOOKEEPER_ADDRESS"); exists { - return strings.Split(value, ",") + address, addrExists := os.LookupEnv("ZOOKEEPER_ADDRESS") + port, portExists := os.LookupEnv("ZOOKEEPER_PORT") + + if addrExists && portExists { + addresses := strings.Split(address, ",") + for i, addr := range addresses { + addresses[i] = addr + ":" + port + } + return addresses } return fallback } diff --git a/pkg/common/kafka/producer.go b/pkg/common/kafka/producer.go index 9a06809b7..06b1e2b4c 100644 --- a/pkg/common/kafka/producer.go +++ b/pkg/common/kafka/producer.go @@ -31,15 +31,14 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/config" ) -const ( - maxRetry = 10 // Maximum number of retries for producer creation -) +const maxRetry = 10 // number of retries -var errEmptyMsg = errors.New("binary msg is empty") +var errEmptyMsg = errors.New("kafka binary msg is empty") +// Producer represents a Kafka producer. type Producer struct { - topic string addr []string + topic string config *sarama.Config producer sarama.SyncProducer } @@ -68,7 +67,7 @@ func NewKafkaProducer(addr []string, topic string) *Producer { // Get Kafka configuration from environment variables or fallback to config file kafkaUsername := getEnvOrConfig("KAFKA_USERNAME", config.Config.Kafka.Username) kafkaPassword := getEnvOrConfig("KAFKA_PASSWORD", config.Config.Kafka.Password) - kafkaAddr := getEnvOrConfig("KAFKA_ADDRESS", addr[0]) // Assuming addr[0] contains address from config + kafkaAddr := getKafkaAddrFromEnv(addr) // Updated to use the new function // Configure SASL authentication if credentials are provided if kafkaUsername != "" && kafkaPassword != "" { @@ -78,7 +77,7 @@ func NewKafkaProducer(addr []string, topic string) *Producer { } // Set the Kafka address - p.addr = []string{kafkaAddr} + p.addr = kafkaAddr // Set up TLS configuration (if required) SetupTLSConfig(p.config) diff --git a/pkg/common/kafka/util.go b/pkg/common/kafka/util.go index da0c5349b..f318ecf73 100644 --- a/pkg/common/kafka/util.go +++ b/pkg/common/kafka/util.go @@ -15,7 +15,9 @@ package kafka import ( + "fmt" "os" + "strings" "github.com/IBM/sarama" @@ -44,3 +46,20 @@ func getEnvOrConfig(envName string, configValue string) string { } return configValue } + +// getKafkaAddrFromEnv returns the Kafka addresses combined from the KAFKA_ADDRESS and KAFKA_PORT environment variables. +// If the environment variables are not set, it returns the fallback value. +func getKafkaAddrFromEnv(fallback []string) []string { + envAddr := os.Getenv("KAFKA_ADDRESS") + envPort := os.Getenv("KAFKA_PORT") + + if envAddr != "" && envPort != "" { + addresses := strings.Split(envAddr, ",") + for i, addr := range addresses { + addresses[i] = fmt.Sprintf("%s:%s", addr, envPort) + } + return addresses + } + + return fallback +} diff --git a/pkg/msgprocessor/options.go b/pkg/msgprocessor/options.go index c17c7cb05..ea383a1af 100644 --- a/pkg/msgprocessor/options.go +++ b/pkg/msgprocessor/options.go @@ -30,10 +30,9 @@ func NewOptions(opts ...OptionsOpt) Options { options[constant.IsOfflinePush] = false options[constant.IsUnreadCount] = false options[constant.IsConversationUpdate] = false - options[constant.IsSenderSync] = false + options[constant.IsSenderSync] = true options[constant.IsNotPrivate] = false options[constant.IsSenderConversationUpdate] = false - options[constant.IsSenderNotificationPush] = false options[constant.IsReactionFromCache] = false for _, opt := range opts { opt(options) @@ -114,12 +113,6 @@ func WithSenderConversationUpdate() OptionsOpt { } } -func WithSenderNotificationPush() OptionsOpt { - return func(options Options) { - options[constant.IsSenderNotificationPush] = true - } -} - func WithReactionFromCache() OptionsOpt { return func(options Options) { options[constant.IsReactionFromCache] = true @@ -174,10 +167,6 @@ func (o Options) IsSenderConversationUpdate() bool { return o.Is(constant.IsSenderConversationUpdate) } -func (o Options) IsSenderNotificationPush() bool { - return o.Is(constant.IsSenderNotificationPush) -} - func (o Options) IsReactionFromCache() bool { return o.Is(constant.IsReactionFromCache) } diff --git a/pkg/rpcclient/msg.go b/pkg/rpcclient/msg.go index 434dc19d0..abad0075a 100644 --- a/pkg/rpcclient/msg.go +++ b/pkg/rpcclient/msg.go @@ -67,6 +67,7 @@ func newContentTypeConf() map[int32]config.NotificationConf { constant.BlackAddedNotification: config.Config.Notification.BlackAdded, constant.BlackDeletedNotification: config.Config.Notification.BlackDeleted, constant.FriendInfoUpdatedNotification: config.Config.Notification.FriendInfoUpdated, + constant.FriendsInfoUpdateNotification: config.Config.Notification.FriendInfoUpdated, //use the same FriendInfoUpdated // conversation constant.ConversationChangeNotification: config.Config.Notification.ConversationChanged, constant.ConversationUnreadNotification: config.Config.Notification.ConversationChanged, @@ -114,6 +115,7 @@ func newSessionTypeConf() map[int32]int32 { constant.BlackAddedNotification: constant.SingleChatType, constant.BlackDeletedNotification: constant.SingleChatType, constant.FriendInfoUpdatedNotification: constant.SingleChatType, + constant.FriendsInfoUpdateNotification: constant.SingleChatType, // conversation constant.ConversationChangeNotification: constant.SingleChatType, constant.ConversationUnreadNotification: constant.SingleChatType, @@ -279,6 +281,7 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s optionsConfig.ReliabilityLevel = constant.UnreliableNotification } options := config.GetOptionsByNotification(optionsConfig) + s.SetOptionsByContentType(ctx, options, contentType) msg.Options = options offlineInfo.Title = title offlineInfo.Desc = desc @@ -297,3 +300,11 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s func (s *NotificationSender) Notification(ctx context.Context, sendID, recvID string, contentType int32, m proto.Message, opts ...NotificationOptions) error { return s.NotificationWithSesstionType(ctx, sendID, recvID, contentType, s.sessionTypeConf[contentType], m, opts...) } + +func (s *NotificationSender) SetOptionsByContentType(_ context.Context, options map[string]bool, contentType int32) { + switch contentType { + case constant.UserStatusChangeNotification: + options[constant.IsSenderSync] = false + default: + } +} diff --git a/pkg/rpcclient/notification/friend.go b/pkg/rpcclient/notification/friend.go index b061a24ae..00759b1b2 100644 --- a/pkg/rpcclient/notification/friend.go +++ b/pkg/rpcclient/notification/friend.go @@ -196,7 +196,12 @@ func (f *FriendNotificationSender) FriendRemarkSetNotification(ctx context.Conte tips.FromToUserID.ToUserID = toUserID return f.Notification(ctx, fromUserID, toUserID, constant.FriendRemarkSetNotification, &tips) } - +func (f *FriendNotificationSender) FriendsInfoUpdateNotification(ctx context.Context, toUserID string, friendIDs []string) error { + tips := sdkws.FriendsInfoUpdateTips{} + tips.FromToUserID.ToUserID = toUserID + tips.FriendIDs = friendIDs + return f.Notification(ctx, toUserID, toUserID, constant.FriendsInfoUpdateNotification, &tips) +} func (f *FriendNotificationSender) BlackAddedNotification(ctx context.Context, req *pbfriend.AddBlackReq) error { tips := sdkws.BlackAddedTips{FromToUserID: &sdkws.FromToUserID{}} tips.FromToUserID.FromUserID = req.OwnerUserID diff --git a/pkg/rpcclient/user.go b/pkg/rpcclient/user.go index de633ee30..451914cd3 100644 --- a/pkg/rpcclient/user.go +++ b/pkg/rpcclient/user.go @@ -64,6 +64,9 @@ func NewUserRpcClient(client discoveryregistry.SvcDiscoveryRegistry) UserRpcClie // GetUsersInfo retrieves information for multiple users based on their user IDs. func (u *UserRpcClient) GetUsersInfo(ctx context.Context, userIDs []string) ([]*sdkws.UserInfo, error) { + if len(userIDs) == 0 { + return []*sdkws.UserInfo{}, nil + } resp, err := u.Client.GetDesignateUsers(ctx, &user.GetDesignateUsersReq{ UserIDs: userIDs, }) diff --git a/scripts/check-all.sh b/scripts/check-all.sh index e1e07bd65..7b53e4a89 100755 --- a/scripts/check-all.sh +++ b/scripts/check-all.sh @@ -30,29 +30,32 @@ OPENIM_VERBOSE=4 openim::log::info "\n# Begin to check all openim service" -# OpenIM status +# Elegant printing function # Elegant printing function print_services_and_ports() { - service_names=("$1[@]") - service_ports=("$2[@]") + local service_names=("$@") + local half_length=$((${#service_names[@]} / 2)) + local service_ports=("${service_names[@]:half_length}") echo "+-------------------------+----------+" echo "| Service Name | Port |" echo "+-------------------------+----------+" - for index in "${!service_names}"; do - printf "| %-23s | %-8s |\n" "${!service_names[$index]}" "${!service_ports[$index]}" + for ((index=0; index < half_length; index++)); do + printf "| %-23s | %-8s |\n" "${service_names[$index]}" "${service_ports[$index]}" done echo "+-------------------------+----------+" } +# Assuming OPENIM_SERVER_NAME_TARGETS and OPENIM_SERVER_PORT_TARGETS are defined +# Similarly for OPENIM_DEPENDENCY_TARGETS and OPENIM_DEPENDENCY_PORT_TARGETS + # Print out services and their ports -print_services_and_ports OPENIM_SERVER_NAME_TARGETS OPENIM_SERVER_PORT_TARGETS +print_services_and_ports "${OPENIM_SERVER_NAME_TARGETS[@]}" "${OPENIM_SERVER_PORT_TARGETS[@]}" # Print out dependencies and their ports -print_services_and_ports OPENIM_DEPENDENCY_TARGETS OPENIM_DEPENDENCY_PORT_TARGETS - +print_services_and_ports "${OPENIM_DEPENDENCY_TARGETS[@]}" "${OPENIM_DEPENDENCY_PORT_TARGETS[@]}" # OpenIM check echo "++ The port being checked: ${OPENIM_SERVER_PORT_LISTARIES[@]}" @@ -89,4 +92,4 @@ else echo "++++ Check all openim service ports successfully !" fi -set -e \ No newline at end of file +set -e diff --git a/scripts/install-im-server.sh b/scripts/install-im-server.sh index 47db34433..a21ae134d 100755 --- a/scripts/install-im-server.sh +++ b/scripts/install-im-server.sh @@ -43,10 +43,20 @@ fi "${OPENIM_ROOT}"/scripts/init-config.sh pushd "${OPENIM_ROOT}" ${DOCKER_COMPOSE_COMMAND} stop -curl https://gitee.com/openimsdk/openim-docker/raw/main/example/full-openim-server-and-chat.yml -o docker-compose.yml +curl https://raw.githubusercontent.com/openimsdk/openim-docker/main/docker-compose.yaml -o docker-compose.yml ${DOCKER_COMPOSE_COMMAND} up -d -sleep 60 + +# Wait for a short period to allow containers to initialize +sleep 30 + +# Check the status of the containers +if ! ${DOCKER_COMPOSE_COMMAND} ps | grep -q 'Up'; then + echo "Error: One or more docker containers failed to start." + ${DOCKER_COMPOSE_COMMAND} logs +fi + +sleep 30 # Keep the original 60-second wait, adjusted for the 10-second check above ${DOCKER_COMPOSE_COMMAND} logs openim-server ${DOCKER_COMPOSE_COMMAND} ps -popd \ No newline at end of file +popd diff --git a/scripts/install/environment.sh b/scripts/install/environment.sh index feb36c3c1..22a0996fc 100755 --- a/scripts/install/environment.sh +++ b/scripts/install/environment.sh @@ -285,7 +285,6 @@ readonly ALERTMANAGER_SEND_RESOLVED=${ALERTMANAGER_SEND_RESOLVED:-"{SEND_RESOLVE ###################### Grafana 配置信息 ###################### def "GRAFANA_PORT" "13000" # Grafana的端口 def "GRAFANA_ADDRESS" "${DOCKER_BRIDGE_GATEWAY}" # Grafana的地址 - ###################### RPC Port Configuration Variables ###################### # For launching multiple programs, just fill in multiple ports separated by commas # For example: @@ -378,8 +377,8 @@ def "CALLBACK_TIMEOUT" "5" # 最长超时时间 def "CALLBACK_FAILED_CONTINUE" "true" # 失败后是否继续 ###################### Prometheus 配置信息 ###################### # 是否启用 Prometheus -readonly PROMETHEUS_ENABLE=${PROMETHEUS_ENABLE:-'false'} -def "PROMETHEUS_URL" "${GRAFANA_ADDRESS}:${GRAFANA_PORT}" +readonly PROMETHEUS_ENABLE=${PROMETHEUS_ENABLE:-'true'} +readonly GRAFANA_URL=${GRAFANA_URL:-"http://${OPENIM_IP}:${GRAFANA_PORT}/"} # Api 服务的 Prometheus 端口 readonly API_PROM_PORT=${API_PROM_PORT:-'20100'} # User 服务的 Prometheus 端口 diff --git a/tools/component/component.go b/tools/component/component.go index 28ea7a2fe..e09fa11e6 100644 --- a/tools/component/component.go +++ b/tools/component/component.go @@ -16,6 +16,7 @@ package main import ( "context" + "errors" "flag" "fmt" "net" @@ -222,8 +223,8 @@ func checkMinio() (string, error) { defer cancel() if minioClient.IsOffline() { - str := fmt.Sprintf("Minio server is offline;%s", str) - return "", ErrComponentStart.Wrap(str) + // str := fmt.Sprintf("Minio server is offline;%s", str) + // return "", ErrComponentStart.Wrap(str) } // Check for localhost in API URL and Minio SignEndpoint @@ -285,10 +286,23 @@ func checkZookeeper() (string, error) { // Connect to Zookeeper str := "the addr is:" + address - c, _, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary + c, eventChan, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary if err != nil { return "", errs.Wrap(errStr(err, str)) } + timeout := time.After(5 * time.Second) + for { + select { + case event := <-eventChan: + if event.State == zk.StateConnected { + fmt.Println("Connected to Zookeeper") + goto Connected + } + case <-timeout: + return "", errs.Wrap(errors.New("timeout waiting for Zookeeper connection"), "Zookeeper Addr: "+strings.Join(config.Config.Zookeeper.ZkAddr, " ")) + } + } +Connected: defer c.Close() // Set authentication if username and password are provided @@ -298,12 +312,6 @@ func checkZookeeper() (string, error) { } } - // Check if Zookeeper is reachable - _, _, err = c.Get("/") - if err != nil { - return "", errs.Wrap(err, str) - } - return str, nil } diff --git a/tools/component/component_test.go b/tools/component/component_test.go index 08ea0fda8..4488c029e 100644 --- a/tools/component/component_test.go +++ b/tools/component/component_test.go @@ -21,22 +21,10 @@ import ( "time" "github.com/redis/go-redis/v9" - "github.com/stretchr/testify/assert" "github.com/openimsdk/open-im-server/v3/pkg/common/config" ) -func TestCheckMysql(t *testing.T) { - err := mockInitCfg() - assert.NoError(t, err, "Initialization should not produce errors") - - err = checkMysql() - if err != nil { - // You might expect an error if MySQL isn't running locally with the mock credentials. - t.Logf("Expected error due to mock configuration: %v", err) - } -} - // Mock for initCfg for testing purpose func mockInitCfg() error { config.Config.Mysql.Username = "root"