diff --git a/.github/workflows/publish-docker-image.yml b/.github/workflows/publish-docker-image.yml index 998a11cf3..4cd3316dd 100644 --- a/.github/workflows/publish-docker-image.yml +++ b/.github/workflows/publish-docker-image.yml @@ -4,42 +4,80 @@ on: push: branches: - release-* - # tags: - # - 'v*' - release: types: [published] - workflow_dispatch: inputs: tag: description: "Tag version to be used for Docker image" required: true - default: "v3.8.0" + default: "v3.8.3" + +env: + GO_VERSION: "1.22" + IMAGE_NAME: "openim-server" + # IMAGE_NAME: ${{ github.event.repository.name }} + DOCKER_BUILDKIT: 1 jobs: - build-and-test: + publish-docker-images: runs-on: ubuntu-latest + if: ${{ !(github.event_name == 'pull_request' && github.event.pull_request.merged == false) }} steps: - - uses: actions/checkout@v4 + - name: Checkout main repository + uses: actions/checkout@v4 with: path: main-repo - # - name: Set up QEMU - # uses: docker/setup-qemu-action@v3.3.0 + - name: Set up QEMU + uses: docker/setup-qemu-action@v3.3.0 - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3.8.0 + id: buildx + uses: docker/setup-buildx-action@v3 + with: + driver-opts: network=host - - name: Build Docker image - id: build - uses: docker/build-push-action@v5 + - name: Extract metadata for Docker + id: meta + uses: docker/metadata-action@v5.6.0 with: - context: ./main-repo - load: true - tags: "openim/openim-server:local" - cache-from: type=gha,scope=build - cache-to: type=gha,mode=max,scope=build + images: | + ${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }} + ghcr.io/${{ github.repository_owner }}/${{ env.IMAGE_NAME }} + registry.cn-hangzhou.aliyuncs.com/openimsdk/${{ env.IMAGE_NAME }} + tags: | + type=ref,event=tag + type=schedule + type=ref,event=branch + type=ref,event=pr + type=semver,pattern={{version}} + type=semver,pattern=v{{version}} + type=semver,pattern={{major}}.{{minor}} + type=semver,pattern={{major}} + type=sha + + - name: Install skopeo + run: | + sudo apt-get update && sudo apt-get install -y skopeo + + - name: Build multi-arch images as OCI + run: | + mkdir -p /tmp/oci-image /tmp/docker-cache + + # Build multi-architecture image and save in OCI format + docker buildx build \ + --platform linux/amd64,linux/arm64 \ + --output type=oci,dest=/tmp/oci-image/multi-arch.tar \ + --cache-to type=local,dest=/tmp/docker-cache \ + --cache-from type=gha \ + ./main-repo + + # Use skopeo to convert the amd64 image from OCI format to Docker format and load it + skopeo copy --override-arch amd64 oci-archive:/tmp/oci-image/multi-arch.tar docker-daemon:${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local + + # check image + docker image ls | grep openim - name: Checkout compose repository uses: actions/checkout@v4 @@ -52,11 +90,11 @@ jobs: run: | IP=$(hostname -I | awk '{print $1}') echo "The IP Address is: $IP" - echo "::set-output name=ip::$IP" + echo "ip=$IP" >> $GITHUB_OUTPUT - name: Update .env to use the local image run: | - sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=openim/openim-server:local|' ${{ github.workspace }}/compose-repo/.env + sed -i 's|OPENIM_SERVER_IMAGE=.*|OPENIM_SERVER_IMAGE=${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}:local|' ${{ github.workspace }}/compose-repo/.env sed -i 's|MINIO_EXTERNAL_ADDRESS=.*|MINIO_EXTERNAL_ADDRESS=http://${{ steps.get-ip.outputs.ip }}:10005|' ${{ github.workspace }}/compose-repo/.env - name: Start services using Docker Compose @@ -66,23 +104,34 @@ jobs: docker compose ps - - name: Extract metadata for Docker (tags, labels) - id: meta - uses: docker/metadata-action@v5.6.0 - with: - images: | - openim/openim-server - ghcr.io/openimsdk/openim-server - registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server - tags: | - type=ref,event=tag - type=schedule - type=ref,event=branch - # type=semver,pattern={{version}} - type=semver,pattern=v{{version}} - type=semver,pattern=release-{{raw}} - type=sha - type=raw,value=${{ github.event.inputs.tag }} + # - name: Check openim-server health + # run: | + # timeout=300 + # interval=30 + # elapsed=0 + # while [[ $elapsed -le $timeout ]]; do + # if ! docker exec openim-server mage check; then + # echo "openim-server is not ready, waiting..." + # sleep $interval + # elapsed=$(($elapsed + $interval)) + # else + # echo "Health check successful" + # exit 0 + # fi + # done + # echo "Health check failed after 5 minutes" + # exit 1 + + # - name: Check openim-chat health + # if: success() + # run: | + # if ! docker exec openim-chat mage check; then + # echo "openim-chat check failed" + # exit 1 + # else + # echo "Health check successful" + # exit 0 + # fi - name: Log in to Docker Hub uses: docker/login-action@v3.3.0 @@ -104,22 +153,27 @@ jobs: username: ${{ secrets.ALIREGISTRY_USERNAME }} password: ${{ secrets.ALIREGISTRY_TOKEN }} - - name: Push Docker images - uses: docker/build-push-action@v5 - with: - context: ./main-repo - push: true - platforms: linux/amd64,linux/arm64 - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha,scope=build - cache-to: type=gha,mode=max,scope=build + - name: Push multi-architecture images + if: success() + run: | + docker buildx build \ + --platform linux/amd64,linux/arm64 \ + $(echo "${{ steps.meta.outputs.tags }}" | sed 's/,/ --tag /g' | sed 's/^/--tag /') \ + --cache-from type=local,src=/tmp/docker-cache \ + --push \ + ./main-repo - name: Verify multi-platform support run: | - images=("openim/openim-server" "ghcr.io/openimsdk/openim-server" "registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server") + images=( + "${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}" + "ghcr.io/${{ github.repository_owner }}/${{ env.IMAGE_NAME }}" + "registry.cn-hangzhou.aliyuncs.com/openimsdk/${{ env.IMAGE_NAME }}" + ) + for image in "${images[@]}"; do - for tag in $(echo "${{ steps.meta.outputs.tags }}" | tr ',' '\n'); do + for tag in $(echo "${{ steps.meta.outputs.tags }}" | tr ',' '\n' | cut -d':' -f2); do + echo "Verifying multi-arch support for $image:$tag" manifest=$(docker manifest inspect "$image:$tag" || echo "error") if [[ "$manifest" == "error" ]]; then echo "Manifest not found for $image:$tag" @@ -135,5 +189,6 @@ jobs: echo "Multi-platform support check failed for $image:$tag - missing arm64" exit 1 fi + echo "✅ $image:$tag supports both amd64 and arm64 architectures" done done diff --git a/config/local-cache.yml b/config/local-cache.yml index 06e211ebb..036dfaa17 100644 --- a/config/local-cache.yml +++ b/config/local-cache.yml @@ -1,3 +1,10 @@ +auth: + topic: DELETE_CACHE_AUTH + slotNum: 100 + slotSize: 2000 + successExpire: 300 + failedExpire: 5 + user: topic: DELETE_CACHE_USER slotNum: 100 diff --git a/config/openim-api.yml b/config/openim-api.yml index 103c36f95..df0177d24 100644 --- a/config/openim-api.yml +++ b/config/openim-api.yml @@ -17,3 +17,13 @@ prometheus: ports: # This address can be accessed via a browser grafanaURL: + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 diff --git a/config/openim-msggateway.yml b/config/openim-msggateway.yml index 812df90f7..8ac07faae 100644 --- a/config/openim-msggateway.yml +++ b/config/openim-msggateway.yml @@ -26,3 +26,20 @@ longConnSvr: websocketMaxMsgLen: 4096 # WebSocket connection handshake timeout in seconds websocketTimeout: 10 + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-msgtransfer.yml b/config/openim-msgtransfer.yml index 52d6a805e..c6ea06803 100644 --- a/config/openim-msgtransfer.yml +++ b/config/openim-msgtransfer.yml @@ -6,3 +6,20 @@ prometheus: # List of ports that Prometheus listens on; each port corresponds to an instance of monitoring. Ensure these are managed accordingly # It will only take effect when autoSetPorts is set to false. ports: + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-push.yml b/config/openim-push.yml index 5db5b541a..58784f13d 100644 --- a/config/openim-push.yml +++ b/config/openim-push.yml @@ -10,10 +10,26 @@ rpc: # It will only take effect when autoSetPorts is set to false. ports: +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached prometheus: # Enable or disable Prometheus monitoring - enable: true + enable: false # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup # It will only take effect when autoSetPorts is set to false. ports: diff --git a/config/openim-rpc-auth.yml b/config/openim-rpc-auth.yml index c42e556c4..09ceef0f4 100644 --- a/config/openim-rpc-auth.yml +++ b/config/openim-rpc-auth.yml @@ -20,3 +20,20 @@ prometheus: tokenPolicy: # Token validity period, in days expire: 90 + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-rpc-conversation.yml b/config/openim-rpc-conversation.yml index e722ac2b0..1135ffe6e 100644 --- a/config/openim-rpc-conversation.yml +++ b/config/openim-rpc-conversation.yml @@ -16,3 +16,20 @@ prometheus: # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup # It will only take effect when autoSetPorts is set to false. ports: + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-rpc-friend.yml b/config/openim-rpc-friend.yml index e722ac2b0..1135ffe6e 100644 --- a/config/openim-rpc-friend.yml +++ b/config/openim-rpc-friend.yml @@ -16,3 +16,20 @@ prometheus: # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup # It will only take effect when autoSetPorts is set to false. ports: + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-rpc-group.yml b/config/openim-rpc-group.yml index 252f64c28..154b149e1 100644 --- a/config/openim-rpc-group.yml +++ b/config/openim-rpc-group.yml @@ -19,3 +19,20 @@ prometheus: enableHistoryForNewMembers: true + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-rpc-msg.yml b/config/openim-rpc-msg.yml index 7781fb5ca..e05eeb7e4 100644 --- a/config/openim-rpc-msg.yml +++ b/config/openim-rpc-msg.yml @@ -20,3 +20,20 @@ prometheus: # Does sending messages require friend verification friendVerify: false + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/openim-rpc-third.yml b/config/openim-rpc-third.yml index 7169e6c61..f6fbee695 100644 --- a/config/openim-rpc-third.yml +++ b/config/openim-rpc-third.yml @@ -17,6 +17,22 @@ prometheus: # It will only take effect when autoSetPorts is set to false. ports: +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached object: # Use MinIO as object storage, or set to "cos", "oss", "kodo", "aws", while also configuring the corresponding settings diff --git a/config/openim-rpc-user.yml b/config/openim-rpc-user.yml index 337cacd35..c46db7f37 100644 --- a/config/openim-rpc-user.yml +++ b/config/openim-rpc-user.yml @@ -16,3 +16,20 @@ prometheus: # Prometheus listening ports, must be consistent with the number of rpc.ports # It will only take effect when autoSetPorts is set to false. ports: + +ratelimiter: + # Whether to enable rate limiting + enable: false + # WindowSize defines time duration per window + window: 20s + # BucketNum defines bucket number for each window + bucket: 500 + # CPU threshold; valid range 0–1000 (1000 = 100%) + cpuThreshold: 850 + +circuitBreaker: + enable: false + window: 5s # Time window size (seconds) + bucket: 100 # Number of buckets + success: 0.6 # Success rate threshold (0.6 means 60%) + request: 500 # Request threshold; circuit breaker evaluation occurs when reached \ No newline at end of file diff --git a/config/webhooks.yml b/config/webhooks.yml index 9fd3eb339..c1645e7c8 100644 --- a/config/webhooks.yml +++ b/config/webhooks.yml @@ -41,6 +41,9 @@ afterSendGroupMsg: attentionIds: [] # See beforeSendSingleMsg comment. deniedTypes: [] +afterMsgSaveDB: + enable: false + timeout: 5 afterUserOnline: enable: false timeout: 5 diff --git a/docker-compose.yml b/docker-compose.yml index 233348619..92eadf237 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -63,7 +63,12 @@ services: restart: always sysctls: net.core.somaxconn: 1024 - command: redis-server /usr/local/redis/config/redis.conf --requirepass openIM123 --appendonly yes + command: > + redis-server + --requirepass openIM123 + --appendonly yes + --aof-use-rdb-preamble yes + --save "" networks: - openim @@ -208,7 +213,6 @@ services: # Defines which listener is used for inter-broker communication within the Kafka cluster KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "INTERNAL" - # Authentication configuration variables - comment out to disable auth # KAFKA_USERNAME: "openIM" # KAFKA_PASSWORD: "openIM123" @@ -262,14 +266,14 @@ services: networks: - openim - openim-admin-front: - image: ${OPENIM_ADMIN_FRONT_IMAGE} - container_name: openim-admin-front - restart: always - ports: - - "11002:80" - networks: - - openim + # openim-admin-front: + # image: ${OPENIM_ADMIN_FRONT_IMAGE} + # container_name: openim-admin-front + # restart: always + # ports: + # - "11002:80" + # networks: + # - openim prometheus: image: ${PROMETHEUS_IMAGE} diff --git a/go.mod b/go.mod index ba1f4adfb..f8c5fa73c 100644 --- a/go.mod +++ b/go.mod @@ -12,8 +12,8 @@ require ( github.com/gorilla/websocket v1.5.1 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/mitchellh/mapstructure v1.5.0 - github.com/openimsdk/protocol v0.0.73-alpha.12 - github.com/openimsdk/tools v0.0.50-alpha.103 + github.com/openimsdk/protocol v0.0.73-alpha.17 + github.com/openimsdk/tools v0.0.50-alpha.105 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 github.com/stretchr/testify v1.10.0 @@ -135,6 +135,7 @@ require ( github.com/leodido/go-urn v1.4.0 // indirect github.com/lestrrat-go/strftime v1.0.6 // indirect github.com/lithammer/shortuuid v3.0.0+incompatible // indirect + github.com/lufia/plan9stats v0.0.0-20230110061619-bbe2e5e100de // indirect github.com/magefile/mage v1.15.0 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mailru/easyjson v0.7.7 // indirect @@ -151,6 +152,7 @@ require ( github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/pierrec/lz4/v4 v4.1.21 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect github.com/prometheus/client_model v0.5.0 // indirect github.com/prometheus/common v0.45.0 // indirect github.com/prometheus/procfs v0.12.0 // indirect @@ -160,6 +162,8 @@ require ( github.com/sagikazarmark/locafero v0.4.0 // indirect github.com/sagikazarmark/slog-shim v0.1.0 // indirect github.com/sercand/kuberesolver/v6 v6.0.1 // indirect + github.com/shirou/gopsutil/v3 v3.24.5 // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.11.0 // indirect github.com/spf13/cast v1.6.0 // indirect diff --git a/go.sum b/go.sum index 830d14c34..ea874f8bc 100644 --- a/go.sum +++ b/go.sum @@ -303,6 +303,8 @@ github.com/likexian/gokit v0.25.13 h1:p2Uw3+6fGG53CwdU2Dz0T6bOycdb2+bAFAa3ymwWVk github.com/likexian/gokit v0.25.13/go.mod h1:qQhEWFBEfqLCO3/vOEo2EDKd+EycekVtUK4tex+l2H4= 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/lufia/plan9stats v0.0.0-20230110061619-bbe2e5e100de h1:V53FWzU6KAZVi1tPp5UIsMoUWJ2/PNwYIDXnu7QuBCE= +github.com/lufia/plan9stats v0.0.0-20230110061619-bbe2e5e100de/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE= github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg= github.com/magefile/mage v1.15.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= @@ -347,10 +349,10 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y= github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM= github.com/openimsdk/gomake v0.0.15-alpha.11 h1:PQudYDRESYeYlUYrrLLJhYIlUPO5x7FAx+o5El9U/Bw= github.com/openimsdk/gomake v0.0.15-alpha.11/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= -github.com/openimsdk/protocol v0.0.73-alpha.12 h1:2NYawXeHChYUeSme6QJ9pOLh+Empce2WmwEtbP4JvKk= -github.com/openimsdk/protocol v0.0.73-alpha.12/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw= -github.com/openimsdk/tools v0.0.50-alpha.103 h1:jYvI86cWiVu8a8iw1panw+pwIiStuUHF76h3fxA6ESI= -github.com/openimsdk/tools v0.0.50-alpha.103/go.mod h1:qCExFBqXpQBMzZck3XGIFwivBayAn2KNqB3WAd++IJw= +github.com/openimsdk/protocol v0.0.73-alpha.17 h1:ddo0QMns1GVwAmrPIPlAQ7uKmThAYLnOt+CIOgLsJyE= +github.com/openimsdk/protocol v0.0.73-alpha.17/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw= +github.com/openimsdk/tools v0.0.50-alpha.105 h1:axuCvKXhxY2RGLhpMMFNgBtE0B65T2Sr1JDW3UD9nBs= +github.com/openimsdk/tools v0.0.50-alpha.105/go.mod h1:x9i/e+WJFW4tocy6RNJQ9NofQiP3KJ1Y576/06TqOG4= github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM= github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs= github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= @@ -361,6 +363,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig= +github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g= github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U= github.com/prometheus/client_golang v1.18.0 h1:HzFfmkOzH5Q8L8G+kSJKUx5dtG87sewO+FoDDqP5Tbk= @@ -397,6 +401,12 @@ github.com/sercand/kuberesolver/v6 v6.0.1 h1:XZUTA0gy/lgDYp/UhEwv7Js24F1j8NJ833Q github.com/sercand/kuberesolver/v6 v6.0.1/go.mod h1:C0tsTuRMONSY+Xf7pv7RMW1/JlewY1+wS8SZE+1lf1s= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= +github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI= +github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk= +github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= +github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= +github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU= +github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k= github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo= github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0= github.com/spf13/afero v1.11.0 h1:WJQKhtpdm3v2IzqG8VMqrr6Rf3UYpEF239Jy9wNepM8= @@ -548,6 +558,7 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/internal/api/conversation.go b/internal/api/conversation.go index 5a191c0ec..78affee44 100644 --- a/internal/api/conversation.go +++ b/internal/api/conversation.go @@ -76,3 +76,7 @@ func (o *ConversationApi) GetPinnedConversationIDs(c *gin.Context) { func (o *ConversationApi) UpdateConversationsByUser(c *gin.Context) { a2r.Call(c, conversation.ConversationClient.UpdateConversationsByUser, o.Client) } + +func (o *ConversationApi) DeleteConversations(c *gin.Context) { + a2r.Call(c, conversation.ConversationClient.DeleteConversations, o.Client) +} diff --git a/internal/api/msg.go b/internal/api/msg.go index a134dfdb3..06fd14936 100644 --- a/internal/api/msg.go +++ b/internal/api/msg.go @@ -219,6 +219,8 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM data = &apistruct.CustomElem{} case constant.MarkdownText: data = &apistruct.MarkdownTextElem{} + case constant.Quote: + data = &apistruct.QuoteElem{} case constant.OANotification: data = &apistruct.OANotificationElem{} req.SessionType = constant.NotificationChatType diff --git a/internal/api/ratelimit.go b/internal/api/ratelimit.go new file mode 100644 index 000000000..0bbf973de --- /dev/null +++ b/internal/api/ratelimit.go @@ -0,0 +1,83 @@ +package api + +import ( + "fmt" + "math" + "net/http" + "strconv" + "time" + + "github.com/gin-gonic/gin" + + "github.com/openimsdk/tools/apiresp" + "github.com/openimsdk/tools/errs" + "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/stability/ratelimit" + "github.com/openimsdk/tools/stability/ratelimit/bbr" +) + +type RateLimiter struct { + Enable bool `yaml:"enable"` + Window time.Duration `yaml:"window"` // time duration per window + Bucket int `yaml:"bucket"` // bucket number for each window + CPUThreshold int64 `yaml:"cpuThreshold"` // CPU threshold; valid range 0–1000 (1000 = 100%) +} + +func RateLimitMiddleware(config *RateLimiter) gin.HandlerFunc { + if !config.Enable { + return func(c *gin.Context) { + c.Next() + } + } + + limiter := bbr.NewBBRLimiter( + bbr.WithWindow(config.Window), + bbr.WithBucket(config.Bucket), + bbr.WithCPUThreshold(config.CPUThreshold), + ) + + return func(c *gin.Context) { + status := limiter.Stat() + + c.Header("X-BBR-CPU", strconv.FormatInt(status.CPU, 10)) + c.Header("X-BBR-MinRT", strconv.FormatInt(status.MinRt, 10)) + c.Header("X-BBR-MaxPass", strconv.FormatInt(status.MaxPass, 10)) + c.Header("X-BBR-MaxInFlight", strconv.FormatInt(status.MaxInFlight, 10)) + c.Header("X-BBR-InFlight", strconv.FormatInt(status.InFlight, 10)) + + done, err := limiter.Allow() + if err != nil { + + c.Header("X-RateLimit-Policy", "BBR") + c.Header("Retry-After", calculateBBRRetryAfter(status)) + c.Header("X-RateLimit-Limit", strconv.FormatInt(status.MaxInFlight, 10)) + c.Header("X-RateLimit-Remaining", "0") // There is no concept of remaining quota in BBR. + + fmt.Println("rate limited:", err, "path:", c.Request.URL.Path) + log.ZWarn(c, "rate limited", err, "path", c.Request.URL.Path) + c.AbortWithStatus(http.StatusTooManyRequests) + apiresp.GinError(c, errs.NewCodeError(http.StatusTooManyRequests, "too many requests, please try again later")) + return + } + + c.Next() + done(ratelimit.DoneInfo{}) + } +} + +func calculateBBRRetryAfter(status bbr.Stat) string { + loadRatio := float64(status.CPU) / float64(status.CPU) + + if loadRatio < 0.8 { + return "1" + } + if loadRatio < 0.95 { + return "2" + } + + backoff := 1 + int64(math.Pow(loadRatio-0.95, 2)*50) + if backoff > 5 { + backoff = 5 + } + return strconv.FormatInt(backoff, 10) +} diff --git a/internal/api/router.go b/internal/api/router.go index 818e144cc..81787f9a9 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -97,6 +97,18 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf case BestSpeed: r.Use(gzip.Gzip(gzip.BestSpeed)) } + + // Use rate limiter middleware + if cfg.API.RateLimiter.Enable { + rl := &RateLimiter{ + Enable: cfg.API.RateLimiter.Enable, + Window: cfg.API.RateLimiter.Window, + Bucket: cfg.API.RateLimiter.Bucket, + CPUThreshold: cfg.API.RateLimiter.CPUThreshold, + } + r.Use(RateLimitMiddleware(rl)) + } + if config.Standalone() { r.Use(func(c *gin.Context) { c.Set(authverify.CtxAdminUserIDsKey, cfg.Share.IMAdminUser.UserIDs) @@ -277,6 +289,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf conversationGroup.POST("/get_owner_conversation", c.GetOwnerConversation) conversationGroup.POST("/get_not_notify_conversation_ids", c.GetNotNotifyConversationIDs) conversationGroup.POST("/get_pinned_conversation_ids", c.GetPinnedConversationIDs) + conversationGroup.POST("/delete_conversations", c.DeleteConversations) conversationGroup.POST("/update_conversations_by_user", c.UpdateConversationsByUser) } diff --git a/internal/msggateway/client.go b/internal/msggateway/client.go index bdb62aece..7b9f4bc0e 100644 --- a/internal/msggateway/client.go +++ b/internal/msggateway/client.go @@ -70,6 +70,7 @@ type Client struct { UserID string `json:"userID"` IsBackground bool `json:"isBackground"` SDKType string `json:"sdkType"` + SDKVersion string `json:"sdkVersion"` Encoder Encoder ctx *UserConnContext longConnServer LongConnServer @@ -97,6 +98,7 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer c.closedErr = nil c.token = ctx.GetToken() c.SDKType = ctx.GetSDKType() + c.SDKVersion = ctx.GetSDKVersion() c.hbCtx, c.hbCancel = context.WithCancel(c.ctx) c.subLock = new(sync.Mutex) if c.subUserIDs != nil { diff --git a/internal/msggateway/constant.go b/internal/msggateway/constant.go index 1e7ab3bb7..3959e1138 100644 --- a/internal/msggateway/constant.go +++ b/internal/msggateway/constant.go @@ -28,6 +28,7 @@ const ( BackgroundStatus = "isBackground" SendResponse = "isMsgResp" SDKType = "sdkType" + SDKVersion = "sdkVersion" ) const ( diff --git a/internal/msggateway/context.go b/internal/msggateway/context.go index d73a96df4..37b5a7cdc 100644 --- a/internal/msggateway/context.go +++ b/internal/msggateway/context.go @@ -15,12 +15,13 @@ package msggateway import ( - "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" "net/http" "net/url" "strconv" "time" + "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" + "github.com/openimsdk/protocol/constant" "github.com/openimsdk/tools/utils/encrypt" "github.com/openimsdk/tools/utils/stringutil" @@ -140,6 +141,10 @@ func (c *UserConnContext) GetToken() string { return c.Req.URL.Query().Get(Token) } +func (c *UserConnContext) GetSDKVersion() string { + return c.Req.URL.Query().Get(SDKVersion) +} + func (c *UserConnContext) GetCompression() bool { compression, exists := c.Query(Compression) if exists && compression == GzipCompressionProtocol { diff --git a/internal/msggateway/ws_server.go b/internal/msggateway/ws_server.go index bc7a2fa5f..d490cc8b9 100644 --- a/internal/msggateway/ws_server.go +++ b/internal/msggateway/ws_server.go @@ -13,6 +13,7 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" "github.com/openimsdk/open-im-server/v3/pkg/rpccache" pbAuth "github.com/openimsdk/protocol/auth" + "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/mcontext" "github.com/go-playground/validator/v10" @@ -64,6 +65,8 @@ type WsServer struct { webhookClient *webhook.Client userClient *rpcli.UserClient authClient *rpcli.AuthClient + + ready atomic.Bool } type kickHandler struct { @@ -93,6 +96,8 @@ func (ws *WsServer) SetDiscoveryRegistry(ctx context.Context, disCov discovery.C ws.authClient = rpcli.NewAuthClient(authConn) ws.MessageHandler = NewGrpcHandler(ws.validate, rpcli.NewMsgClient(msgConn), rpcli.NewPushMsgServiceClient(pushConn)) ws.disCov = disCov + + ws.ready.Store(true) return nil } @@ -254,6 +259,10 @@ func (ws *WsServer) registerClient(client *Client) { oldClients []*Client ) oldClients, userOK, clientOK = ws.clients.Get(client.UserID, client.PlatformID) + + log.ZInfo(client.ctx, "registerClient", "userID", client.UserID, "platformID", client.PlatformID, + "sdkVersion", client.SDKVersion) + if !userOK { ws.clients.Set(client.UserID, client) log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID) @@ -453,6 +462,11 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { // Create a new connection context connContext := newContext(w, r) + if !ws.ready.Load() { + httpError(connContext, errs.New("ws server not ready")) + return + } + // Check if the current number of online user connections exceeds the maximum limit if ws.onlineUserConnNum.Load() >= ws.wsMaxConnNum { // If it exceeds the maximum connection number, return an error via HTTP and stop processing @@ -469,6 +483,11 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { return } + if ws.authClient == nil { + httpError(connContext, errs.New("auth client is not initialized")) + return + } + // Call the authentication client to parse the Token obtained from the context resp, err := ws.authClient.ParseToken(connContext, connContext.GetToken()) if err != nil { diff --git a/internal/msgtransfer/callback.go b/internal/msgtransfer/callback.go index f0d439779..575efb4ae 100644 --- a/internal/msgtransfer/callback.go +++ b/internal/msgtransfer/callback.go @@ -51,37 +51,24 @@ func GetContent(msg *sdkws.MsgData) string { } } -func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) { - if msg.ContentType == constant.Typing { - return - } - +func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterMsgSaveDB(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) { if !filterAfterMsg(msg, after) { return } - cbReq := &cbapi.CallbackAfterSendSingleMsgReq{ - CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand), - RecvID: msg.RecvID, - } - mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg)) -} - -func (mc *OnlineHistoryMongoConsumerHandler) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *sdkws.MsgData) { - if msg.ContentType == constant.Typing { - return - } - - if !filterAfterMsg(msg, after) { - return + cbReq := &cbapi.CallbackAfterMsgSaveDBReq{ + CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterMsgSaveDBCommand), } - cbReq := &cbapi.CallbackAfterSendGroupMsgReq{ - CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand), - GroupID: msg.GroupID, + switch msg.SessionType { + case constant.SingleChatType, constant.NotificationChatType: + cbReq.RecvID = msg.RecvID + case constant.ReadGroupChatType: + cbReq.GroupID = msg.GroupID + default: } - mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg)) + mc.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterMsgSaveDBResp{}, after, buildKeyMsgDataQuery(msg)) } func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string { diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go index 8b212774a..8af585ce3 100644 --- a/internal/msgtransfer/online_history_msg_handler.go +++ b/internal/msgtransfer/online_history_msg_handler.go @@ -18,6 +18,7 @@ import ( "context" "encoding/json" "errors" + "github.com/openimsdk/tools/mq" "sync" diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index 8611af7ea..147bd37b0 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -15,7 +15,6 @@ package msgtransfer import ( - "github.com/openimsdk/protocol/constant" "github.com/openimsdk/tools/mq" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" @@ -57,7 +56,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String()) err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq) if err != nil { - log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) + log.ZError(ctx, "batch data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) prommetrics.MsgInsertMongoFailedCounter.Inc() } else { prommetrics.MsgInsertMongoSuccessCounter.Inc() @@ -65,12 +64,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) } for _, msgData := range msgFromMQ.MsgData { - switch msgData.SessionType { - case constant.SingleChatType: - mc.webhookAfterSendSingleMsg(ctx, &mc.config.WebhooksConfig.AfterSendSingleMsg, msgData) - case constant.ReadGroupChatType: - mc.webhookAfterSendGroupMsg(ctx, &mc.config.WebhooksConfig.AfterSendGroupMsg, msgData) - } + mc.webhookAfterMsgSaveDB(ctx, &mc.config.WebhooksConfig.AfterMsgSaveDB, msgData) } //var seqs []int64 diff --git a/internal/rpc/auth/auth.go b/internal/rpc/auth/auth.go index 7a8607164..a78a714ca 100644 --- a/internal/rpc/auth/auth.go +++ b/internal/rpc/auth/auth.go @@ -18,10 +18,13 @@ import ( "context" "errors" + "github.com/openimsdk/open-im-server/v3/pkg/common/convert" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo" "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" + "github.com/openimsdk/open-im-server/v3/pkg/localcache" + "github.com/openimsdk/open-im-server/v3/pkg/rpccache" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/open-im-server/v3/pkg/common/config" @@ -46,6 +49,7 @@ import ( type authServer struct { pbauth.UnimplementedAuthServer authDatabase controller.AuthDatabase + AuthLocalCache *rpccache.AuthLocalCache RegisterCenter discovery.Conn config *Config userClient *rpcli.UserClient @@ -53,11 +57,12 @@ type authServer struct { } type Config struct { - RpcConfig config.Auth - RedisConfig config.Redis - MongoConfig config.Mongo - Share config.Share - Discovery config.Discovery + RpcConfig config.Auth + RedisConfig config.Redis + MongoConfig config.Mongo + Share config.Share + LocalCacheConfig config.LocalCache + Discovery config.Discovery } func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error { @@ -78,12 +83,19 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg } token = mcache.NewTokenCacheModel(mc, config.RpcConfig.TokenPolicy.Expire) } else { - token = redis2.NewTokenCacheModel(rdb, config.RpcConfig.TokenPolicy.Expire) + token = redis2.NewTokenCacheModel(rdb, &config.LocalCacheConfig, config.RpcConfig.TokenPolicy.Expire) } userConn, err := client.GetConn(ctx, config.Discovery.RpcService.User) if err != nil { return err } + authConn, err := client.GetConn(ctx, config.Discovery.RpcService.Auth) + if err != nil { + return err + } + + localcache.InitLocalCache(&config.LocalCacheConfig) + pbauth.RegisterAuthServer(server, &authServer{ RegisterCenter: client, authDatabase: controller.NewAuthDatabase( @@ -93,9 +105,10 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg config.Share.MultiLogin, config.Share.IMAdminUser.UserIDs, ), - config: config, - userClient: rpcli.NewUserClient(userConn), - adminUserIDs: config.Share.IMAdminUser.UserIDs, + AuthLocalCache: rpccache.NewAuthLocalCache(rpcli.NewAuthClient(authConn), &config.LocalCacheConfig, rdb), + config: config, + userClient: rpcli.NewUserClient(userConn), + adminUserIDs: config.Share.IMAdminUser.UserIDs, }) return nil } @@ -121,6 +134,7 @@ func (s *authServer) GetAdminToken(ctx context.Context, req *pbauth.GetAdminToke } prommetrics.UserLoginCounter.Inc() + resp.Token = token resp.ExpireTimeSeconds = s.config.RpcConfig.TokenPolicy.Expire * 24 * 60 * 60 return &resp, nil @@ -151,20 +165,34 @@ func (s *authServer) GetUserToken(ctx context.Context, req *pbauth.GetUserTokenR if err != nil { return nil, err } + resp.Token = token resp.ExpireTimeSeconds = s.config.RpcConfig.TokenPolicy.Expire * 24 * 60 * 60 return &resp, nil } +func (s *authServer) GetExistingToken(ctx context.Context, req *pbauth.GetExistingTokenReq) (*pbauth.GetExistingTokenResp, error) { + m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID)) + if err != nil { + return nil, err + } + + return &pbauth.GetExistingTokenResp{ + TokenStates: convert.TokenMapDB2Pb(m), + }, nil +} + func (s *authServer) parseToken(ctx context.Context, tokensString string) (claims *tokenverify.Claims, err error) { claims, err = tokenverify.GetClaimFromToken(tokensString, authverify.Secret(s.config.Share.Secret)) if err != nil { return nil, err } - m, err := s.authDatabase.GetTokensWithoutError(ctx, claims.UserID, claims.PlatformID) + + m, err := s.AuthLocalCache.GetExistingToken(ctx, claims.UserID, claims.PlatformID) if err != nil { return nil, err } + if len(m) == 0 { isAdmin := authverify.CheckUserIsAdmin(ctx, claims.UserID) if isAdmin { diff --git a/internal/rpc/conversation/conversation.go b/internal/rpc/conversation/conversation.go index a2b72ddfb..562a76d82 100644 --- a/internal/rpc/conversation/conversation.go +++ b/internal/rpc/conversation/conversation.go @@ -37,6 +37,7 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/protocol/constant" pbconversation "github.com/openimsdk/protocol/conversation" + "github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/errs" @@ -132,6 +133,7 @@ func (c *conversationServer) GetConversation(ctx context.Context, req *pbconvers return resp, nil } +// Deprecated: Use `GetConversations` instead. func (c *conversationServer) GetSortedConversationList(ctx context.Context, req *pbconversation.GetSortedConversationListReq) (resp *pbconversation.GetSortedConversationListResp, err error) { if err := authverify.CheckAccess(ctx, req.UserID); err != nil { return nil, err @@ -183,9 +185,21 @@ func (c *conversationServer) GetSortedConversationList(ctx context.Context, req conversation_isPinTime := make(map[int64]string) conversation_notPinTime := make(map[int64]string) + for _, v := range conversations { conversationID := v.ConversationID - time := conversationMsg[conversationID].MsgInfo.LatestMsgRecvTime + var time int64 + if _, ok := conversationMsg[conversationID]; ok { + time = conversationMsg[conversationID].MsgInfo.LatestMsgRecvTime + } else { + conversationMsg[conversationID] = &pbconversation.ConversationElem{ + ConversationID: conversationID, + IsPinned: v.IsPinned, + MsgInfo: nil, + } + time = v.CreateTime.UnixMilli() + } + conversationMsg[conversationID].RecvMsgOpt = v.RecvMsgOpt if v.IsPinned { conversationMsg[conversationID].IsPinned = v.IsPinned @@ -782,7 +796,7 @@ func (c *conversationServer) ClearUserConversationMsg(ctx context.Context, req * } latestMsgDestructTime := time.UnixMilli(req.Timestamp) for i, conversation := range conversations { - if conversation.IsMsgDestruct == false || conversation.MsgDestructTime == 0 { + if !conversation.IsMsgDestruct || conversation.MsgDestructTime == 0 { continue } seq, err := c.msgClient.GetLastMessageSeqByTime(ctx, conversation.ConversationID, req.Timestamp-(conversation.MsgDestructTime*1000)) @@ -822,3 +836,53 @@ func (c *conversationServer) setConversationMinSeqAndLatestMsgDestructTime(ctx c c.conversationNotificationSender.ConversationChangeNotification(ctx, ownerUserID, []string{conversationID}) return nil } + +func (c *conversationServer) DeleteConversations(ctx context.Context, req *pbconversation.DeleteConversationsReq) (resp *pbconversation.DeleteConversationsResp, err error) { + if err := authverify.CheckAccess(ctx, req.OwnerUserID); err != nil { + return nil, err + } + if req.NeedDeleteTime == 0 && len(req.ConversationIDs) == 0 { + return nil, errs.ErrArgs.WrapMsg("need_delete_time or conversationIDs need be set") + } + + if req.NeedDeleteTime != 0 && len(req.ConversationIDs) != 0 { + return nil, errs.ErrArgs.WrapMsg("need_delete_time and conversationIDs cannot both be set") + } + + var needDeleteConversationIDs []string + + if len(req.ConversationIDs) == 0 { + deleteTimeThreshold := time.Now().AddDate(0, 0, -int(req.NeedDeleteTime)).UnixMilli() + conversationIDs, err := c.conversationDatabase.GetConversationIDs(ctx, req.OwnerUserID) + if err != nil { + return nil, err + } + latestMsgs, err := c.msgClient.GetLastMessage(ctx, &msg.GetLastMessageReq{ + UserID: req.OwnerUserID, + ConversationIDs: conversationIDs, + }) + if err != nil { + return nil, err + } + + for conversationID, msg := range latestMsgs.Msgs { + if msg.SendTime < deleteTimeThreshold { + needDeleteConversationIDs = append(needDeleteConversationIDs, conversationID) + } + } + + if len(needDeleteConversationIDs) == 0 { + return &pbconversation.DeleteConversationsResp{}, nil + } + } else { + needDeleteConversationIDs = req.ConversationIDs + } + + if err := c.conversationDatabase.DeleteUsersConversations(ctx, req.OwnerUserID, needDeleteConversationIDs); err != nil { + return nil, err + } + + // c.conversationNotificationSender.ConversationDeleteNotification(ctx, req.OwnerUserID, needDeleteConversationIDs) + + return &pbconversation.DeleteConversationsResp{}, nil +} diff --git a/internal/rpc/conversation/notification.go b/internal/rpc/conversation/notification.go index 6e865ba42..aa7b7d227 100644 --- a/internal/rpc/conversation/notification.go +++ b/internal/rpc/conversation/notification.go @@ -73,3 +73,12 @@ func (c *ConversationNotificationSender) ConversationUnreadChangeNotification( c.Notification(ctx, userID, userID, constant.ConversationUnreadNotification, tips) } + +func (c *ConversationNotificationSender) ConversationDeleteNotification(ctx context.Context, userID string, conversationIDs []string) { + tips := &sdkws.ConversationDeleteTips{ + UserID: userID, + ConversationIDs: conversationIDs, + } + + c.Notification(ctx, userID, userID, constant.ConversationDeleteNotification, tips) +} diff --git a/internal/rpc/group/sync.go b/internal/rpc/group/sync.go index 47b7a8306..92c7a60ce 100644 --- a/internal/rpc/group/sync.go +++ b/internal/rpc/group/sync.go @@ -2,6 +2,7 @@ package group import ( "context" + "errors" "github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" "github.com/openimsdk/open-im-server/v3/pkg/authverify" @@ -11,6 +12,7 @@ import ( "github.com/openimsdk/protocol/constant" pbgroup "github.com/openimsdk/protocol/group" "github.com/openimsdk/protocol/sdkws" + "github.com/openimsdk/tools/log" ) const versionSyncLimit = 500 @@ -170,19 +172,26 @@ func (g *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup. func (g *groupServer) BatchGetIncrementalGroupMember(ctx context.Context, req *pbgroup.BatchGetIncrementalGroupMemberReq) (*pbgroup.BatchGetIncrementalGroupMemberResp, error) { var num int resp := make(map[string]*pbgroup.GetIncrementalGroupMemberResp) + for _, memberReq := range req.ReqList { if _, ok := resp[memberReq.GroupID]; ok { continue } memberResp, err := g.GetIncrementalGroupMember(ctx, memberReq) if err != nil { + if errors.Is(err, servererrs.ErrDismissedAlready) { + log.ZWarn(ctx, "Failed to get incremental group member", err, "groupID", memberReq.GroupID, "request", memberReq) + continue + } return nil, err } + resp[memberReq.GroupID] = memberResp num += len(memberResp.Insert) + len(memberResp.Update) + len(memberResp.Delete) if num >= versionSyncLimit { break } } + return &pbgroup.BatchGetIncrementalGroupMemberResp{RespList: resp}, nil } diff --git a/internal/rpc/msg/callback.go b/internal/rpc/msg/callback.go index 2c00efd43..49fb04477 100644 --- a/internal/rpc/msg/callback.go +++ b/internal/rpc/msg/callback.go @@ -16,8 +16,10 @@ package msg import ( "context" + "encoding/base64" "encoding/json" + "github.com/openimsdk/open-im-server/v3/pkg/apistruct" "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" "github.com/openimsdk/tools/errs" @@ -28,6 +30,7 @@ import ( "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/utils/datautil" + "github.com/openimsdk/tools/utils/stringutil" "google.golang.org/protobuf/proto" ) @@ -87,19 +90,19 @@ func (m *msgServer) webhookBeforeSendSingleMsg(ctx context.Context, before *conf } // Move to msgtransfer -// func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) { -// if msg.MsgData.ContentType == constant.Typing { -// return -// } -// if !filterAfterMsg(msg, after) { -// return -// } -// cbReq := &cbapi.CallbackAfterSendSingleMsgReq{ -// CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand), -// RecvID: msg.MsgData.RecvID, -// } -// m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData)) -// } +func (m *msgServer) webhookAfterSendSingleMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) { + if msg.MsgData.ContentType == constant.Typing { + return + } + if !filterAfterMsg(msg, after) { + return + } + cbReq := &cbapi.CallbackAfterSendSingleMsgReq{ + CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendSingleMsgCommand), + RecvID: msg.MsgData.RecvID, + } + m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendSingleMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData)) +} func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq) error { return webhook.WithCondition(ctx, before, func(ctx context.Context) error { @@ -121,21 +124,20 @@ func (m *msgServer) webhookBeforeSendGroupMsg(ctx context.Context, before *confi }) } -// Move to msgtransfer -// func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) { -// if msg.MsgData.ContentType == constant.Typing { -// return -// } -// if !filterAfterMsg(msg, after) { -// return -// } -// cbReq := &cbapi.CallbackAfterSendGroupMsgReq{ -// CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand), -// GroupID: msg.MsgData.GroupID, -// } - -// m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData)) -// } +func (m *msgServer) webhookAfterSendGroupMsg(ctx context.Context, after *config.AfterConfig, msg *pbchat.SendMsgReq) { + if msg.MsgData.ContentType == constant.Typing { + return + } + if !filterAfterMsg(msg, after) { + return + } + cbReq := &cbapi.CallbackAfterSendGroupMsgReq{ + CommonCallbackReq: toCommonCallback(ctx, msg, cbapi.CallbackAfterSendGroupMsgCommand), + GroupID: msg.MsgData.GroupID, + } + + m.webhookClient.AsyncPostWithQuery(ctx, cbReq.GetCallbackCommand(), cbReq, &cbapi.CallbackAfterSendGroupMsgResp{}, after, buildKeyMsgDataQuery(msg.MsgData)) +} func (m *msgServer) webhookBeforeMsgModify(ctx context.Context, before *config.BeforeConfig, msg *pbchat.SendMsgReq, beforeMsgData **sdkws.MsgData) error { return webhook.WithCondition(ctx, before, func(ctx context.Context) error { @@ -204,14 +206,14 @@ func (m *msgServer) webhookAfterRevokeMsg(ctx context.Context, after *config.Aft m.webhookClient.AsyncPost(ctx, callbackReq.GetCallbackCommand(), callbackReq, &cbapi.CallbackAfterRevokeMsgResp{}, after) } -// func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string { -// keyMsgData := apistruct.KeyMsgData{ -// SendID: msg.SendID, -// RecvID: msg.RecvID, -// GroupID: msg.GroupID, -// } - -// return map[string]string{ -// webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)), -// } -// } +func buildKeyMsgDataQuery(msg *sdkws.MsgData) map[string]string { + keyMsgData := apistruct.KeyMsgData{ + SendID: msg.SendID, + RecvID: msg.RecvID, + GroupID: msg.GroupID, + } + + return map[string]string{ + webhook.Key: base64.StdEncoding.EncodeToString(stringutil.StructToJsonBytes(keyMsgData)), + } +} diff --git a/internal/rpc/msg/send.go b/internal/rpc/msg/send.go index d97905bff..18ad9cc56 100644 --- a/internal/rpc/msg/send.go +++ b/internal/rpc/msg/send.go @@ -86,7 +86,7 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq, go m.setConversationAtInfo(ctx, req.MsgData) } - // m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req) + m.webhookAfterSendGroupMsg(ctx, &m.config.WebhooksConfig.AfterSendGroupMsg, req) prommetrics.GroupChatMsgProcessSuccessCounter.Inc() resp = &pbmsg.SendMsgResp{} @@ -194,7 +194,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq return nil, err } - // m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req) + m.webhookAfterSendSingleMsg(ctx, &m.config.WebhooksConfig.AfterSendSingleMsg, req) prommetrics.SingleChatMsgProcessSuccessCounter.Inc() return &pbmsg.SendMsgResp{ ServerMsgID: req.MsgData.ServerMsgID, diff --git a/pkg/callbackstruct/constant.go b/pkg/callbackstruct/constant.go index bbc34e71f..ef7cb502f 100644 --- a/pkg/callbackstruct/constant.go +++ b/pkg/callbackstruct/constant.go @@ -66,4 +66,5 @@ const ( CallbackAfterCreateSingleChatConversationsCommand = "callbackAfterCreateSingleChatConversationsCommand" CallbackBeforeCreateGroupChatConversationsCommand = "callbackBeforeCreateGroupChatConversationsCommand" CallbackAfterCreateGroupChatConversationsCommand = "callbackAfterCreateGroupChatConversationsCommand" + CallbackAfterMsgSaveDBCommand = "callbackAfterMsgSaveDBCommand" ) diff --git a/pkg/callbackstruct/message.go b/pkg/callbackstruct/message.go index 902fa6110..ef8a8bb28 100644 --- a/pkg/callbackstruct/message.go +++ b/pkg/callbackstruct/message.go @@ -103,3 +103,13 @@ type CallbackSingleMsgReadReq struct { type CallbackSingleMsgReadResp struct { CommonCallbackResp } + +type CallbackAfterMsgSaveDBReq struct { + CommonCallbackReq + RecvID string `json:"recvID"` + GroupID string `json:"groupID"` +} + +type CallbackAfterMsgSaveDBResp struct { + CommonCallbackResp +} diff --git a/pkg/common/cmd/api.go b/pkg/common/cmd/api.go index 7b7dbc89b..1356f90d6 100644 --- a/pkg/common/cmd/api.go +++ b/pkg/common/cmd/api.go @@ -81,6 +81,9 @@ func (a *ApiCmd) runE() error { } return startrpc.Start( a.ctx, &a.apiConfig.Discovery, + nil, + nil, + // &a.apiConfig.API.RateLimiter, &prometheus, a.apiConfig.API.Api.ListenIP, "", a.apiConfig.API.Prometheus.AutoSetPorts, diff --git a/pkg/common/cmd/auth.go b/pkg/common/cmd/auth.go index 6e59b6dc6..a6a92ef14 100644 --- a/pkg/common/cmd/auth.go +++ b/pkg/common/cmd/auth.go @@ -40,6 +40,7 @@ func NewAuthRpcCmd() *AuthRpcCmd { config.RedisConfigFileName: &authConfig.RedisConfig, config.MongodbConfigFileName: &authConfig.MongoConfig, config.ShareFileName: &authConfig.Share, + config.LocalCacheConfigFileName: &authConfig.LocalCacheConfig, config.DiscoveryConfigFilename: &authConfig.Discovery, } ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap)) @@ -56,7 +57,7 @@ func (a *AuthRpcCmd) Exec() error { } func (a *AuthRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.authConfig.Discovery, &a.authConfig.RpcConfig.Prometheus, a.authConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.authConfig.Discovery, &a.authConfig.RpcConfig.CircuitBreaker, &a.authConfig.RpcConfig.RateLimiter, &a.authConfig.RpcConfig.Prometheus, a.authConfig.RpcConfig.RPC.ListenIP, a.authConfig.RpcConfig.RPC.RegisterIP, a.authConfig.RpcConfig.RPC.AutoSetPorts, a.authConfig.RpcConfig.RPC.Ports, a.Index(), a.authConfig.Discovery.RpcService.Auth, nil, a.authConfig, []string{ diff --git a/pkg/common/cmd/conversation.go b/pkg/common/cmd/conversation.go index 428c442da..f1ae0c59c 100644 --- a/pkg/common/cmd/conversation.go +++ b/pkg/common/cmd/conversation.go @@ -58,7 +58,7 @@ func (a *ConversationRpcCmd) Exec() error { } func (a *ConversationRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.conversationConfig.Discovery, &a.conversationConfig.RpcConfig.Prometheus, a.conversationConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.conversationConfig.Discovery, &a.conversationConfig.RpcConfig.CircuitBreaker, &a.conversationConfig.RpcConfig.RateLimiter, &a.conversationConfig.RpcConfig.Prometheus, a.conversationConfig.RpcConfig.RPC.ListenIP, a.conversationConfig.RpcConfig.RPC.RegisterIP, a.conversationConfig.RpcConfig.RPC.AutoSetPorts, a.conversationConfig.RpcConfig.RPC.Ports, a.Index(), a.conversationConfig.Discovery.RpcService.Conversation, &a.conversationConfig.NotificationConfig, a.conversationConfig, []string{ diff --git a/pkg/common/cmd/cron_task.go b/pkg/common/cmd/cron_task.go index af2cbbee9..c666bd021 100644 --- a/pkg/common/cmd/cron_task.go +++ b/pkg/common/cmd/cron_task.go @@ -56,6 +56,8 @@ func (a *CronTaskCmd) runE() error { var prometheus config.Prometheus return startrpc.Start( a.ctx, &a.cronTaskConfig.Discovery, + nil, + nil, &prometheus, "", "", true, diff --git a/pkg/common/cmd/friend.go b/pkg/common/cmd/friend.go index dd850cf17..661da86c7 100644 --- a/pkg/common/cmd/friend.go +++ b/pkg/common/cmd/friend.go @@ -58,7 +58,7 @@ func (a *FriendRpcCmd) Exec() error { } func (a *FriendRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.relationConfig.Discovery, &a.relationConfig.RpcConfig.Prometheus, a.relationConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.relationConfig.Discovery, &a.relationConfig.RpcConfig.CircuitBreaker, &a.relationConfig.RpcConfig.RateLimiter, &a.relationConfig.RpcConfig.Prometheus, a.relationConfig.RpcConfig.RPC.ListenIP, a.relationConfig.RpcConfig.RPC.RegisterIP, a.relationConfig.RpcConfig.RPC.AutoSetPorts, a.relationConfig.RpcConfig.RPC.Ports, a.Index(), a.relationConfig.Discovery.RpcService.Friend, &a.relationConfig.NotificationConfig, a.relationConfig, []string{ diff --git a/pkg/common/cmd/group.go b/pkg/common/cmd/group.go index 7a599077f..daff7b499 100644 --- a/pkg/common/cmd/group.go +++ b/pkg/common/cmd/group.go @@ -59,7 +59,7 @@ func (a *GroupRpcCmd) Exec() error { } func (a *GroupRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.groupConfig.Discovery, &a.groupConfig.RpcConfig.Prometheus, a.groupConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.groupConfig.Discovery, &a.groupConfig.RpcConfig.CircuitBreaker, &a.groupConfig.RpcConfig.RateLimiter, &a.groupConfig.RpcConfig.Prometheus, a.groupConfig.RpcConfig.RPC.ListenIP, a.groupConfig.RpcConfig.RPC.RegisterIP, a.groupConfig.RpcConfig.RPC.AutoSetPorts, a.groupConfig.RpcConfig.RPC.Ports, a.Index(), a.groupConfig.Discovery.RpcService.Group, &a.groupConfig.NotificationConfig, a.groupConfig, []string{ diff --git a/pkg/common/cmd/msg.go b/pkg/common/cmd/msg.go index c4049be05..845175cd9 100644 --- a/pkg/common/cmd/msg.go +++ b/pkg/common/cmd/msg.go @@ -59,7 +59,7 @@ func (a *MsgRpcCmd) Exec() error { } func (a *MsgRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.msgConfig.Discovery, &a.msgConfig.RpcConfig.Prometheus, a.msgConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.msgConfig.Discovery, &a.msgConfig.RpcConfig.CircuitBreaker, &a.msgConfig.RpcConfig.RateLimiter, &a.msgConfig.RpcConfig.Prometheus, a.msgConfig.RpcConfig.RPC.ListenIP, a.msgConfig.RpcConfig.RPC.RegisterIP, a.msgConfig.RpcConfig.RPC.AutoSetPorts, a.msgConfig.RpcConfig.RPC.Ports, a.Index(), a.msgConfig.Discovery.RpcService.Msg, &a.msgConfig.NotificationConfig, a.msgConfig, []string{ diff --git a/pkg/common/cmd/msg_gateway.go b/pkg/common/cmd/msg_gateway.go index 83cfb6272..4ca899a5e 100644 --- a/pkg/common/cmd/msg_gateway.go +++ b/pkg/common/cmd/msg_gateway.go @@ -61,6 +61,8 @@ func (m *MsgGatewayCmd) runE() error { var prometheus config.Prometheus return startrpc.Start( m.ctx, &m.msgGatewayConfig.Discovery, + &m.msgGatewayConfig.MsgGateway.CircuitBreaker, + &m.msgGatewayConfig.MsgGateway.RateLimiter, &prometheus, rpc.ListenIP, rpc.RegisterIP, rpc.AutoSetPorts, diff --git a/pkg/common/cmd/msg_transfer.go b/pkg/common/cmd/msg_transfer.go index fe6c27e54..81f40ac9d 100644 --- a/pkg/common/cmd/msg_transfer.go +++ b/pkg/common/cmd/msg_transfer.go @@ -62,6 +62,8 @@ func (m *MsgTransferCmd) runE() error { var prometheus config.Prometheus return startrpc.Start( m.ctx, &m.msgTransferConfig.Discovery, + &m.msgTransferConfig.MsgTransfer.CircuitBreaker, + &m.msgTransferConfig.MsgTransfer.RateLimiter, &prometheus, "", "", true, diff --git a/pkg/common/cmd/push.go b/pkg/common/cmd/push.go index 7cd3c481e..bcb34ed2c 100644 --- a/pkg/common/cmd/push.go +++ b/pkg/common/cmd/push.go @@ -60,7 +60,7 @@ func (a *PushRpcCmd) Exec() error { } func (a *PushRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.pushConfig.Discovery, &a.pushConfig.RpcConfig.Prometheus, a.pushConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.pushConfig.Discovery, &a.pushConfig.RpcConfig.CircuitBreaker, &a.pushConfig.RpcConfig.RateLimiter, &a.pushConfig.RpcConfig.Prometheus, a.pushConfig.RpcConfig.RPC.ListenIP, a.pushConfig.RpcConfig.RPC.RegisterIP, a.pushConfig.RpcConfig.RPC.AutoSetPorts, a.pushConfig.RpcConfig.RPC.Ports, a.Index(), a.pushConfig.Discovery.RpcService.Push, &a.pushConfig.NotificationConfig, a.pushConfig, []string{ diff --git a/pkg/common/cmd/third.go b/pkg/common/cmd/third.go index e567234e4..39d731e0d 100644 --- a/pkg/common/cmd/third.go +++ b/pkg/common/cmd/third.go @@ -58,7 +58,7 @@ func (a *ThirdRpcCmd) Exec() error { } func (a *ThirdRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.thirdConfig.Discovery, &a.thirdConfig.RpcConfig.Prometheus, a.thirdConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.thirdConfig.Discovery, &a.thirdConfig.RpcConfig.CircuitBreaker, &a.thirdConfig.RpcConfig.RateLimiter, &a.thirdConfig.RpcConfig.Prometheus, a.thirdConfig.RpcConfig.RPC.ListenIP, a.thirdConfig.RpcConfig.RPC.RegisterIP, a.thirdConfig.RpcConfig.RPC.AutoSetPorts, a.thirdConfig.RpcConfig.RPC.Ports, a.Index(), a.thirdConfig.Discovery.RpcService.Third, &a.thirdConfig.NotificationConfig, a.thirdConfig, []string{ diff --git a/pkg/common/cmd/user.go b/pkg/common/cmd/user.go index 190f6f892..12abe1bde 100644 --- a/pkg/common/cmd/user.go +++ b/pkg/common/cmd/user.go @@ -59,7 +59,7 @@ func (a *UserRpcCmd) Exec() error { } func (a *UserRpcCmd) runE() error { - return startrpc.Start(a.ctx, &a.userConfig.Discovery, &a.userConfig.RpcConfig.Prometheus, a.userConfig.RpcConfig.RPC.ListenIP, + return startrpc.Start(a.ctx, &a.userConfig.Discovery, &a.userConfig.RpcConfig.CircuitBreaker, &a.userConfig.RpcConfig.RateLimiter, &a.userConfig.RpcConfig.Prometheus, a.userConfig.RpcConfig.RPC.ListenIP, a.userConfig.RpcConfig.RPC.RegisterIP, a.userConfig.RpcConfig.RPC.AutoSetPorts, a.userConfig.RpcConfig.RPC.Ports, a.Index(), a.userConfig.Discovery.RpcService.User, &a.userConfig.NotificationConfig, a.userConfig, []string{ diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index 63716838a..695684d15 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -43,6 +43,7 @@ type CacheConfig struct { } type LocalCache struct { + Auth CacheConfig `yaml:"auth"` User CacheConfig `yaml:"user"` Group CacheConfig `yaml:"group"` Friend CacheConfig `yaml:"friend"` @@ -142,6 +143,23 @@ type API struct { Ports []int `yaml:"ports"` GrafanaURL string `yaml:"grafanaURL"` } `yaml:"prometheus"` + + RateLimiter RateLimiter `yaml:"rateLimiter"` +} + +type RateLimiter struct { + Enable bool `yaml:"enable"` + Window time.Duration `yaml:"window"` + Bucket int `yaml:"bucket"` + CPUThreshold int64 `yaml:"cpuThreshold"` +} + +type CircuitBreaker struct { + Enable bool `yaml:"enable"` + Window time.Duration `yaml:"window"` + Bucket int `yaml:"bucket"` + Success float64 `yaml:"success"` + Request int64 `yaml:"request"` } type CronTask struct { @@ -216,6 +234,8 @@ type MsgGateway struct { WebsocketMaxMsgLen int `yaml:"websocketMaxMsgLen"` WebsocketTimeout int `yaml:"websocketTimeout"` } `yaml:"longConnSvr"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type MsgTransfer struct { @@ -224,6 +244,8 @@ type MsgTransfer struct { AutoSetPorts bool `yaml:"autoSetPorts"` Ports []int `yaml:"ports"` } `yaml:"prometheus"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Push struct { @@ -254,7 +276,9 @@ type Push struct { BadgeCount bool `yaml:"badgeCount"` Production bool `yaml:"production"` } `yaml:"iosPush"` - FullUserCache bool `yaml:"fullUserCache"` + FullUserCache bool `yaml:"fullUserCache"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Auth struct { @@ -263,28 +287,38 @@ type Auth struct { TokenPolicy struct { Expire int64 `yaml:"expire"` } `yaml:"tokenPolicy"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Conversation struct { - RPC RPC `yaml:"rpc"` - Prometheus Prometheus `yaml:"prometheus"` + RPC RPC `yaml:"rpc"` + Prometheus Prometheus `yaml:"prometheus"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Friend struct { - RPC RPC `yaml:"rpc"` - Prometheus Prometheus `yaml:"prometheus"` + RPC RPC `yaml:"rpc"` + Prometheus Prometheus `yaml:"prometheus"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Group struct { - RPC RPC `yaml:"rpc"` - Prometheus Prometheus `yaml:"prometheus"` - EnableHistoryForNewMembers bool `yaml:"enableHistoryForNewMembers"` + RPC RPC `yaml:"rpc"` + Prometheus Prometheus `yaml:"prometheus"` + EnableHistoryForNewMembers bool `yaml:"enableHistoryForNewMembers"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Msg struct { - RPC RPC `yaml:"rpc"` - Prometheus Prometheus `yaml:"prometheus"` - FriendVerify bool `yaml:"friendVerify"` + RPC RPC `yaml:"rpc"` + Prometheus Prometheus `yaml:"prometheus"` + FriendVerify bool `yaml:"friendVerify"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Third struct { @@ -297,6 +331,8 @@ type Third struct { Kodo Kodo `yaml:"kodo"` Aws Aws `yaml:"aws"` } `yaml:"object"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type Cos struct { BucketURL string `yaml:"bucketURL"` @@ -335,8 +371,10 @@ type Aws struct { } type User struct { - RPC RPC `yaml:"rpc"` - Prometheus Prometheus `yaml:"prometheus"` + RPC RPC `yaml:"rpc"` + Prometheus Prometheus `yaml:"prometheus"` + RateLimiter RateLimiter `yaml:"rateLimiter"` + CircuitBreaker CircuitBreaker `yaml:"circuitBreaker"` } type RPC struct { @@ -435,6 +473,7 @@ type Webhooks struct { BeforeSendGroupMsg BeforeConfig `yaml:"beforeSendGroupMsg"` BeforeMsgModify BeforeConfig `yaml:"beforeMsgModify"` AfterSendGroupMsg AfterConfig `yaml:"afterSendGroupMsg"` + AfterMsgSaveDB AfterConfig `yaml:"afterMsgSaveDB"` AfterUserOnline AfterConfig `yaml:"afterUserOnline"` AfterUserOffline AfterConfig `yaml:"afterUserOffline"` AfterUserKickOff AfterConfig `yaml:"afterUserKickOff"` diff --git a/pkg/common/convert/auth.go b/pkg/common/convert/auth.go new file mode 100644 index 000000000..a0f00c8fe --- /dev/null +++ b/pkg/common/convert/auth.go @@ -0,0 +1,25 @@ +package convert + +func TokenMapDB2Pb(tokenMapDB map[string]int) map[string]int32 { + if tokenMapDB == nil { + return nil + } + + tokenMapPB := make(map[string]int32, len(tokenMapDB)) + for k, v := range tokenMapDB { + tokenMapPB[k] = int32(v) + } + return tokenMapPB +} + +func TokenMapPb2DB(tokenMapPB map[string]int32) map[string]int { + if tokenMapPB == nil { + return nil + } + + tokenMapDB := make(map[string]int, len(tokenMapPB)) + for k, v := range tokenMapPB { + tokenMapDB[k] = int(v) + } + return tokenMapDB +} \ No newline at end of file diff --git a/pkg/common/startrpc/circuitbreaker.go b/pkg/common/startrpc/circuitbreaker.go new file mode 100644 index 000000000..060a3aa8e --- /dev/null +++ b/pkg/common/startrpc/circuitbreaker.go @@ -0,0 +1,107 @@ +package startrpc + +import ( + "context" + "time" + + "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/stability/circuitbreaker" + "github.com/openimsdk/tools/stability/circuitbreaker/sre" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type CircuitBreaker struct { + Enable bool `yaml:"enable"` + Success float64 `yaml:"success"` // success rate threshold (0.0-1.0) + Request int64 `yaml:"request"` // request threshold + Bucket int `yaml:"bucket"` // number of buckets + Window time.Duration `yaml:"window"` // time window for statistics +} + +func NewCircuitBreaker(config *CircuitBreaker) circuitbreaker.CircuitBreaker { + if !config.Enable { + return nil + } + + return sre.NewSREBraker( + sre.WithWindow(config.Window), + sre.WithBucket(config.Bucket), + sre.WithSuccess(config.Success), + sre.WithRequest(config.Request), + ) +} + +func UnaryCircuitBreakerInterceptor(breaker circuitbreaker.CircuitBreaker) grpc.ServerOption { + if breaker == nil { + return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) { + return handler(ctx, req) + }) + } + + return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) { + if err := breaker.Allow(); err != nil { + log.ZWarn(ctx, "rpc circuit breaker open", err, "method", info.FullMethod) + return nil, status.Error(codes.Unavailable, "service unavailable due to circuit breaker") + } + + resp, err = handler(ctx, req) + + if err != nil { + if st, ok := status.FromError(err); ok { + switch st.Code() { + case codes.OK: + breaker.MarkSuccess() + case codes.InvalidArgument, codes.NotFound, codes.AlreadyExists, codes.PermissionDenied: + breaker.MarkSuccess() + default: + breaker.MarkFailed() + } + } else { + breaker.MarkFailed() + } + } else { + breaker.MarkSuccess() + } + + return resp, err + + }) +} + +func StreamCircuitBreakerInterceptor(breaker circuitbreaker.CircuitBreaker) grpc.ServerOption { + if breaker == nil { + return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + return handler(srv, ss) + }) + } + + return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + if err := breaker.Allow(); err != nil { + log.ZWarn(ss.Context(), "rpc circuit breaker open", err, "method", info.FullMethod) + return status.Error(codes.Unavailable, "service unavailable due to circuit breaker") + } + + err := handler(srv, ss) + + if err != nil { + if st, ok := status.FromError(err); ok { + switch st.Code() { + case codes.OK: + breaker.MarkSuccess() + case codes.InvalidArgument, codes.NotFound, codes.AlreadyExists, codes.PermissionDenied: + breaker.MarkSuccess() + default: + breaker.MarkFailed() + } + } else { + breaker.MarkFailed() + } + } else { + breaker.MarkSuccess() + } + + return err + }) +} diff --git a/pkg/common/startrpc/ratelimit.go b/pkg/common/startrpc/ratelimit.go new file mode 100644 index 000000000..1c2ac8eae --- /dev/null +++ b/pkg/common/startrpc/ratelimit.go @@ -0,0 +1,70 @@ +package startrpc + +import ( + "context" + "time" + + "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/stability/ratelimit" + "github.com/openimsdk/tools/stability/ratelimit/bbr" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type RateLimiter struct { + Enable bool + Window time.Duration + Bucket int + CPUThreshold int64 +} + +func NewRateLimiter(config *RateLimiter) ratelimit.Limiter { + if !config.Enable { + return nil + } + + return bbr.NewBBRLimiter( + bbr.WithWindow(config.Window), + bbr.WithBucket(config.Bucket), + bbr.WithCPUThreshold(config.CPUThreshold), + ) +} + +func UnaryRateLimitInterceptor(limiter ratelimit.Limiter) grpc.ServerOption { + if limiter == nil { + return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) { + return handler(ctx, req) + }) + } + + return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) { + done, err := limiter.Allow() + if err != nil { + log.ZWarn(ctx, "rpc rate limited", err, "method", info.FullMethod) + return nil, status.Errorf(codes.ResourceExhausted, "rpc request rate limit exceeded: %v, please try again later", err) + } + + defer done(ratelimit.DoneInfo{}) + return handler(ctx, req) + }) +} + +func StreamRateLimitInterceptor(limiter ratelimit.Limiter) grpc.ServerOption { + if limiter == nil { + return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + return handler(srv, ss) + }) + } + + return grpc.ChainStreamInterceptor(func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + done, err := limiter.Allow() + if err != nil { + log.ZWarn(ss.Context(), "rpc rate limited", err, "method", info.FullMethod) + return status.Errorf(codes.ResourceExhausted, "rpc request rate limit exceeded: %v, please try again later", err) + } + defer done(ratelimit.DoneInfo{}) + + return handler(srv, ss) + }) +} diff --git a/pkg/common/startrpc/start.go b/pkg/common/startrpc/start.go index 9715f2aac..247d13b6b 100644 --- a/pkg/common/startrpc/start.go +++ b/pkg/common/startrpc/start.go @@ -47,7 +47,7 @@ func init() { prommetrics.RegistryAll() } -func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP, +func Start[T any](ctx context.Context, disc *conf.Discovery, circuitBreakerConfig *conf.CircuitBreaker, rateLimiterConfig *conf.RateLimiter, prometheusConfig *conf.Prometheus, listenIP, registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T, watchConfigNames []string, watchServiceNames []string, rpcFn func(ctx context.Context, config T, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error, @@ -84,6 +84,45 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c } } + if circuitBreakerConfig != nil && circuitBreakerConfig.Enable { + cb := &CircuitBreaker{ + Enable: circuitBreakerConfig.Enable, + Success: circuitBreakerConfig.Success, + Request: circuitBreakerConfig.Request, + Bucket: circuitBreakerConfig.Bucket, + Window: circuitBreakerConfig.Window, + } + + breaker := NewCircuitBreaker(cb) + + options = append(options, + UnaryCircuitBreakerInterceptor(breaker), + StreamCircuitBreakerInterceptor(breaker), + ) + + log.ZInfo(ctx, "RPC circuit breaker enabled", + "service", rpcRegisterName, + "window", circuitBreakerConfig.Window, + "bucket", circuitBreakerConfig.Bucket, + "success", circuitBreakerConfig.Success, + "requestThreshold", circuitBreakerConfig.Request) + } + + if rateLimiterConfig != nil && rateLimiterConfig.Enable { + limiter := NewRateLimiter((*RateLimiter)(rateLimiterConfig)) + + options = append(options, + UnaryRateLimitInterceptor(limiter), + StreamRateLimitInterceptor(limiter), + ) + + log.ZInfo(ctx, "RPC rate limiter enabled", + "service", rpcRegisterName, + "window", rateLimiterConfig.Window, + "bucket", rateLimiterConfig.Bucket, + "cpuThreshold", rateLimiterConfig.CPUThreshold) + } + registerIP, err := network.GetRpcRegisterIP(registerIP) if err != nil { return err @@ -123,7 +162,7 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c go func() { sigs := make(chan os.Signal, 1) - signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL) + signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT) select { case <-ctx.Done(): return diff --git a/pkg/common/storage/cache/conversation.go b/pkg/common/storage/cache/conversation.go index ac3011107..d5ab2264f 100644 --- a/pkg/common/storage/cache/conversation.go +++ b/pkg/common/storage/cache/conversation.go @@ -16,6 +16,7 @@ package cache import ( "context" + relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" ) @@ -57,7 +58,7 @@ type ConversationCache interface { GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache DelConversationNotNotifyMessageUserIDs(userIDs ...string) ConversationCache - DelConversationPinnedMessageUserIDs(userIDs ...string) ConversationCache + DelUserPinnedConversations(userIDs ...string) ConversationCache DelConversationVersionUserIDs(userIDs ...string) ConversationCache FindMaxConversationUserVersion(ctx context.Context, userID string) (*relationtb.VersionLog, error) diff --git a/pkg/common/storage/cache/mcache/msg_cache.go b/pkg/common/storage/cache/mcache/msg_cache.go index 3846be3f8..6fd5f80a1 100644 --- a/pkg/common/storage/cache/mcache/msg_cache.go +++ b/pkg/common/storage/cache/mcache/msg_cache.go @@ -24,7 +24,7 @@ var ( func NewMsgCache(cache database.Cache, msgDocDatabase database.Msg) cache.MsgCache { initMemMsgCache.Do(func() { - memMsgCache = lru.NewLayLRU[string, *model.MsgInfoModel](1024*8, time.Hour, time.Second*10, localcache.EmptyTarget{}, nil) + memMsgCache = lru.NewLazyLRU[string, *model.MsgInfoModel](1024*8, time.Hour, time.Second*10, localcache.EmptyTarget{}, nil) }) return &msgCache{ cache: cache, diff --git a/pkg/common/storage/cache/redis/conversation.go b/pkg/common/storage/cache/redis/conversation.go index 3453b570d..16d67322b 100644 --- a/pkg/common/storage/cache/redis/conversation.go +++ b/pkg/common/storage/cache/redis/conversation.go @@ -253,7 +253,7 @@ func (c *ConversationRedisCache) DelConversationNotNotifyMessageUserIDs(userIDs return cache } -func (c *ConversationRedisCache) DelConversationPinnedMessageUserIDs(userIDs ...string) cache.ConversationCache { +func (c *ConversationRedisCache) DelUserPinnedConversations(userIDs ...string) cache.ConversationCache { cache := c.CloneConversationCache() for _, userID := range userIDs { cache.AddKeys(c.getPinnedConversationIDsKey(userID)) diff --git a/pkg/common/storage/cache/redis/token.go b/pkg/common/storage/cache/redis/token.go index c74ccce66..3a9967e90 100644 --- a/pkg/common/storage/cache/redis/token.go +++ b/pkg/common/storage/cache/redis/token.go @@ -2,13 +2,16 @@ package redis import ( "context" + "encoding/json" "strconv" "sync" "time" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" "github.com/openimsdk/tools/errs" + "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/utils/datautil" "github.com/redis/go-redis/v9" ) @@ -16,16 +19,26 @@ import ( type tokenCache struct { rdb redis.UniversalClient accessExpire time.Duration + localCache *config.LocalCache } -func NewTokenCacheModel(rdb redis.UniversalClient, accessExpire int64) cache.TokenModel { - c := &tokenCache{rdb: rdb} +func NewTokenCacheModel(rdb redis.UniversalClient, localCache *config.LocalCache, accessExpire int64) cache.TokenModel { + c := &tokenCache{rdb: rdb, localCache: localCache} c.accessExpire = c.getExpireTime(accessExpire) return c } func (c *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error { - return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), token, flag).Err()) + key := cachekey.GetTokenKey(userID, platformID) + if err := c.rdb.HSet(ctx, key, token, flag).Err(); err != nil { + return errs.Wrap(err) + } + + if c.localCache != nil { + c.removeLocalTokenCache(ctx, key) + } + + return nil } // SetTokenFlagEx set token and flag with expire time @@ -37,6 +50,11 @@ func (c *tokenCache) SetTokenFlagEx(ctx context.Context, userID string, platform if err := c.rdb.Expire(ctx, key, c.accessExpire).Err(); err != nil { return errs.Wrap(err) } + + if c.localCache != nil { + c.removeLocalTokenCache(ctx, key) + } + return nil } @@ -106,7 +124,17 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla for k, v := range m { mm[k] = v } - return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err()) + + err := c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err() + if err != nil { + return errs.Wrap(err) + } + + if c.localCache != nil { + c.removeLocalTokenCache(ctx, cachekey.GetTokenKey(userID, platformID)) + } + + return nil } func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error { @@ -124,11 +152,23 @@ func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[st }); err != nil { return err } + + if c.localCache != nil { + c.removeLocalTokenCache(ctx, keys...) + } return nil } func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error { - return errs.Wrap(c.rdb.HDel(ctx, cachekey.GetTokenKey(userID, platformID), fields...).Err()) + key := cachekey.GetTokenKey(userID, platformID) + if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil { + return errs.Wrap(err) + } + + if c.localCache != nil { + c.removeLocalTokenCache(ctx, key) + } + return nil } func (c *tokenCache) getExpireTime(t int64) time.Duration { @@ -161,6 +201,11 @@ func (c *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, t return err } + // Remove local cache for the token + if c.localCache != nil { + c.removeLocalTokenCache(ctx, keys...) + } + return nil } @@ -175,5 +220,28 @@ func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, p if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil { return errs.Wrap(err) } + if c.localCache != nil { + c.removeLocalTokenCache(ctx, key) + } return nil } + +func (c *tokenCache) removeLocalTokenCache(ctx context.Context, keys ...string) { + if len(keys) == 0 { + return + } + + topic := c.localCache.Auth.Topic + if topic == "" { + return + } + + data, err := json.Marshal(keys) + if err != nil { + log.ZWarn(ctx, "keys json marshal failed", err, "topic", topic, "keys", keys) + } else { + if err := c.rdb.Publish(ctx, topic, string(data)).Err(); err != nil { + log.ZWarn(ctx, "redis publish cache delete error", err, "topic", topic, "keys", keys) + } + } +} diff --git a/pkg/common/storage/controller/conversation.go b/pkg/common/storage/controller/conversation.go index 27442ca66..d085c7466 100644 --- a/pkg/common/storage/controller/conversation.go +++ b/pkg/common/storage/controller/conversation.go @@ -78,6 +78,8 @@ type ConversationDatabase interface { GetPinnedConversationIDs(ctx context.Context, userID string) ([]string, error) // FindRandConversation finds random conversations based on the specified timestamp and limit. FindRandConversation(ctx context.Context, ts int64, limit int) ([]*relationtb.Conversation, error) + + DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) } func NewConversationDatabase(conversation database.Conversation, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase { @@ -120,7 +122,7 @@ func (c *conversationDatabase) SetUsersConversationFieldTx(ctx context.Context, cache = cache.DelConversationNotNotifyMessageUserIDs(userIDs...) } if _, ok := fieldMap["is_pinned"]; ok { - cache = cache.DelConversationPinnedMessageUserIDs(userIDs...) + cache = cache.DelUserPinnedConversations(userIDs...) } cache = cache.DelConversationVersionUserIDs(haveUserIDs...) } @@ -172,7 +174,7 @@ func (c *conversationDatabase) UpdateUsersConversationField(ctx context.Context, cache = cache.DelConversationNotNotifyMessageUserIDs(userIDs...) } if _, ok := args["is_pinned"]; ok { - cache = cache.DelConversationPinnedMessageUserIDs(userIDs...) + cache = cache.DelUserPinnedConversations(userIDs...) } return cache.ChainExecDel(ctx) } @@ -203,7 +205,7 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat DelUserConversationIDsHash(userIDs...). DelConversationVersionUserIDs(userIDs...). DelConversationNotNotifyMessageUserIDs(notNotifyUserIDs...). - DelConversationPinnedMessageUserIDs(pinnedUserIDs...). + DelUserPinnedConversations(pinnedUserIDs...). ChainExecDel(ctx) } @@ -259,7 +261,7 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs cache := c.cache.CloneConversationCache() cache = cache.DelConversationVersionUserIDs(ownerUserID). DelConversationNotNotifyMessageUserIDs(ownerUserID). - DelConversationPinnedMessageUserIDs(ownerUserID) + DelUserPinnedConversations(ownerUserID) groupIDs := datautil.Distinct(datautil.Filter(conversations, func(e *relationtb.Conversation) (string, bool) { return e.GroupID, e.GroupID != "" @@ -429,3 +431,21 @@ func (c *conversationDatabase) GetPinnedConversationIDs(ctx context.Context, use func (c *conversationDatabase) FindRandConversation(ctx context.Context, ts int64, limit int) ([]*relationtb.Conversation, error) { return c.conversationDB.FindRandConversation(ctx, ts, limit) } + +func (c *conversationDatabase) DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) { + return c.tx.Transaction(ctx, func(ctx context.Context) error { + err = c.conversationDB.DeleteUsersConversations(ctx, userID, conversationIDs) + if err != nil { + return err + } + cache := c.cache.CloneConversationCache() + cache = cache.DelConversations(userID, conversationIDs...). + DelConversationVersionUserIDs(userID). + DelConversationIDs(userID). + DelUserConversationIDsHash(userID). + DelConversationNotNotifyMessageUserIDs(userID). + DelUserPinnedConversations(userID) + + return cache.ChainExecDel(ctx) + }) +} diff --git a/pkg/common/storage/controller/group.go b/pkg/common/storage/controller/group.go index 037ab1c39..24209cd6c 100644 --- a/pkg/common/storage/controller/group.go +++ b/pkg/common/storage/controller/group.go @@ -194,7 +194,6 @@ func (g *groupDatabase) CreateGroup(ctx context.Context, groups []*model.Group, } for _, group := range groups { c = c.DelGroupsInfo(group.GroupID). - DelGroupMembersHash(group.GroupID). DelGroupMembersHash(group.GroupID). DelGroupsMemberNum(group.GroupID). DelGroupMemberIDs(group.GroupID). diff --git a/pkg/common/storage/database/conversation.go b/pkg/common/storage/database/conversation.go index d612dfc2d..562782346 100644 --- a/pkg/common/storage/database/conversation.go +++ b/pkg/common/storage/database/conversation.go @@ -44,4 +44,5 @@ type Conversation interface { GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) FindConversationUserVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error) FindRandConversation(ctx context.Context, ts int64, limit int) ([]*model.Conversation, error) + DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) } diff --git a/pkg/common/storage/database/mgo/conversation.go b/pkg/common/storage/database/mgo/conversation.go index 89f13ea3d..d5ab046aa 100644 --- a/pkg/common/storage/database/mgo/conversation.go +++ b/pkg/common/storage/database/mgo/conversation.go @@ -308,3 +308,20 @@ func (c *ConversationMgo) FindRandConversation(ctx context.Context, ts int64, li } return mongoutil.Aggregate[*model.Conversation](ctx, c.coll, pipeline) } + +func (c *ConversationMgo) DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) { + if len(conversationIDs) == 0 { + return nil + } + return mongoutil.IncrVersion(func() error { + err := mongoutil.DeleteMany(ctx, c.coll, bson.M{"owner_user_id": userID, "conversation_id": bson.M{"$in": conversationIDs}}) + return err + }, func() error { + for _, conversationID := range conversationIDs { + if err := c.version.IncrVersion(ctx, userID, []string{conversationID}, model.VersionStateDelete); err != nil { + return err + } + } + return nil + }) +} diff --git a/pkg/common/storage/database/mgo/msg.go b/pkg/common/storage/database/mgo/msg.go index 975e7a0ef..315f4530b 100644 --- a/pkg/common/storage/database/mgo/msg.go +++ b/pkg/common/storage/database/mgo/msg.go @@ -5,6 +5,11 @@ import ( "fmt" "time" + "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/bson/primitive" + "go.mongodb.org/mongo-driver/mongo" + "go.mongodb.org/mongo-driver/mongo/options" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" "github.com/openimsdk/protocol/constant" @@ -14,10 +19,6 @@ import ( "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/utils/datautil" "github.com/openimsdk/tools/utils/jsonutil" - "go.mongodb.org/mongo-driver/bson" - "go.mongodb.org/mongo-driver/bson/primitive" - "go.mongodb.org/mongo-driver/mongo" - "go.mongodb.org/mongo-driver/mongo/options" ) func NewMsgMongo(db *mongo.Database) (database.Msg, error) { @@ -1154,7 +1155,7 @@ func (m *MsgMgo) findBeforeDocSendTime(ctx context.Context, docID string, limit if err != nil { return 0, 0, err } - for i := len(res) - 1; i > 0; i-- { + for i := len(res) - 1; i >= 0; i-- { v := res[i] if v.Msgs != nil && v.Msgs.Msg != nil && v.Msgs.Msg.SendTime > 0 { return v.Msgs.Msg.Seq, v.Msgs.Msg.SendTime, nil @@ -1169,7 +1170,7 @@ func (m *MsgMgo) findBeforeSendTime(ctx context.Context, conversationID string, limit := int64(-1) if first { first = false - limit = m.model.GetMsgIndex(seq) + limit = m.model.GetLimitForSingleDoc(seq) } docID := m.model.BuildDocIDByIndex(conversationID, i) msgSeq, msgSendTime, err := m.findBeforeDocSendTime(ctx, docID, limit) diff --git a/pkg/common/storage/database/mgo/user.go b/pkg/common/storage/database/mgo/user.go index a08765baf..04ecca1e8 100644 --- a/pkg/common/storage/database/mgo/user.go +++ b/pkg/common/storage/database/mgo/user.go @@ -73,7 +73,7 @@ func (u *UserMgo) TakeNotification(ctx context.Context, level int64) (user []*mo } func (u *UserMgo) TakeGTEAppManagerLevel(ctx context.Context, level int64) (user []*model.User, err error) { - return mongoutil.Find[*model.User](ctx, u.coll, bson.M{"app_manager_level": bson.M{"$gte": level}}) + return mongoutil.Find[*model.User](ctx, u.coll, bson.M{"app_manger_level": bson.M{"$gte": level}}) } func (u *UserMgo) TakeByNickname(ctx context.Context, nickname string) (user []*model.User, err error) { diff --git a/pkg/common/storage/model/msg.go b/pkg/common/storage/model/msg.go index 9d5b56b42..2bb4882f7 100644 --- a/pkg/common/storage/model/msg.go +++ b/pkg/common/storage/model/msg.go @@ -132,6 +132,10 @@ func (*MsgDocModel) GetMsgIndex(seq int64) int64 { return (seq - 1) % singleGocMsgNum } +func (*MsgDocModel) GetLimitForSingleDoc(seq int64) int64 { + return seq % singleGocMsgNum +} + func (*MsgDocModel) indexGen(conversationID string, seqSuffix int64) string { return conversationID + ":" + strconv.FormatInt(seqSuffix, 10) } diff --git a/pkg/localcache/cache.go b/pkg/localcache/cache.go index ba849f892..07d36cf46 100644 --- a/pkg/localcache/cache.go +++ b/pkg/localcache/cache.go @@ -47,15 +47,15 @@ func New[V any](opts ...Option) Cache[V] { if opt.localSlotNum > 0 && opt.localSlotSize > 0 { createSimpleLRU := func() lru.LRU[string, V] { if opt.expirationEvict { - return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict) + return lru.NewExpirationLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict) } else { - return lru.NewLayLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict) + return lru.NewLazyLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict) } } if opt.localSlotNum == 1 { c.local = createSimpleLRU() } else { - c.local = lru.NewSlotLRU[string, V](opt.localSlotNum, LRUStringHash, createSimpleLRU) + c.local = lru.NewSlotLRU(opt.localSlotNum, LRUStringHash, createSimpleLRU) } if opt.linkSlotNum > 0 { c.link = link.New(opt.linkSlotNum) @@ -71,6 +71,8 @@ type cache[V any] struct { } func (c *cache[V]) onEvict(key string, value V) { + _ = value + if c.link != nil { lks := c.link.Del(key) for k := range lks { diff --git a/pkg/localcache/init.go b/pkg/localcache/init.go index d1c16f675..ad339da7c 100644 --- a/pkg/localcache/init.go +++ b/pkg/localcache/init.go @@ -15,10 +15,11 @@ package localcache import ( - "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" "strings" "sync" + + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" ) var ( @@ -32,6 +33,10 @@ func InitLocalCache(localCache *config.LocalCache) { Local config.CacheConfig Keys []string }{ + { + Local: localCache.Auth, + Keys: []string{cachekey.UidPidToken}, + }, { Local: localCache.User, Keys: []string{cachekey.UserInfoKey, cachekey.UserGlobalRecvMsgOptKey}, diff --git a/pkg/localcache/lru/lru_lazy.go b/pkg/localcache/lru/lru_lazy.go index b4f0377a7..4a3db46c9 100644 --- a/pkg/localcache/lru/lru_lazy.go +++ b/pkg/localcache/lru/lru_lazy.go @@ -21,25 +21,25 @@ import ( "github.com/hashicorp/golang-lru/v2/simplelru" ) -type layLruItem[V any] struct { +type lazyLruItem[V any] struct { lock sync.Mutex expires int64 err error value V } -func NewLayLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) *LayLRU[K, V] { - var cb simplelru.EvictCallback[K, *layLruItem[V]] +func NewLazyLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) *LazyLRU[K, V] { + var cb simplelru.EvictCallback[K, *lazyLruItem[V]] if onEvict != nil { - cb = func(key K, value *layLruItem[V]) { + cb = func(key K, value *lazyLruItem[V]) { onEvict(key, value.value) } } - core, err := simplelru.NewLRU[K, *layLruItem[V]](size, cb) + core, err := simplelru.NewLRU[K, *lazyLruItem[V]](size, cb) if err != nil { panic(err) } - return &LayLRU[K, V]{ + return &LazyLRU[K, V]{ core: core, successTTL: successTTL, failedTTL: failedTTL, @@ -47,15 +47,15 @@ func NewLayLRU[K comparable, V any](size int, successTTL, failedTTL time.Duratio } } -type LayLRU[K comparable, V any] struct { +type LazyLRU[K comparable, V any] struct { lock sync.Mutex - core *simplelru.LRU[K, *layLruItem[V]] + core *simplelru.LRU[K, *lazyLruItem[V]] successTTL time.Duration failedTTL time.Duration target Target } -func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) { +func (x *LazyLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) { x.lock.Lock() v, ok := x.core.Get(key) if ok { @@ -68,7 +68,7 @@ func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) { return value, err } } else { - v = &layLruItem[V]{} + v = &lazyLruItem[V]{} x.core.Add(key, v) v.lock.Lock() x.lock.Unlock() @@ -88,15 +88,15 @@ func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) { return v.value, v.err } -func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) { +func (x *LazyLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) { var ( err error once sync.Once ) res := make(map[K]V) - queries := make([]K, 0) - setVs := make(map[K]*layLruItem[V]) + queries := make([]K, 0, len(keys)) + for _, key := range keys { x.lock.Lock() v, ok := x.core.Get(key) @@ -118,14 +118,20 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) } queries = append(queries, key) } - values, err1 := fetch(queries) - if err1 != nil { + + if len(queries) == 0 { + return res, err + } + + values, fetchErr := fetch(queries) + if fetchErr != nil { once.Do(func() { - err = err1 + err = fetchErr }) } + for key, val := range values { - v := &layLruItem[V]{} + v := &lazyLruItem[V]{} v.value = val if err == nil { @@ -135,7 +141,7 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) v.expires = time.Now().Add(x.failedTTL).UnixMilli() x.target.IncrGetFailed() } - setVs[key] = v + x.lock.Lock() x.core.Add(key, v) x.lock.Unlock() @@ -145,29 +151,29 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) return res, err } -//func (x *LayLRU[K, V]) Has(key K) bool { +//func (x *LazyLRU[K, V]) Has(key K) bool { // x.lock.Lock() // defer x.lock.Unlock() // return x.core.Contains(key) //} -func (x *LayLRU[K, V]) Set(key K, value V) { +func (x *LazyLRU[K, V]) Set(key K, value V) { x.lock.Lock() defer x.lock.Unlock() - x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()}) + x.core.Add(key, &lazyLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()}) } -func (x *LayLRU[K, V]) SetHas(key K, value V) bool { +func (x *LazyLRU[K, V]) SetHas(key K, value V) bool { x.lock.Lock() defer x.lock.Unlock() if x.core.Contains(key) { - x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()}) + x.core.Add(key, &lazyLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()}) return true } return false } -func (x *LayLRU[K, V]) Del(key K) bool { +func (x *LazyLRU[K, V]) Del(key K) bool { x.lock.Lock() ok := x.core.Remove(key) x.lock.Unlock() @@ -179,6 +185,6 @@ func (x *LayLRU[K, V]) Del(key K) bool { return ok } -func (x *LayLRU[K, V]) Stop() { +func (x *LazyLRU[K, V]) Stop() { } diff --git a/pkg/localcache/lru/lru_slot.go b/pkg/localcache/lru/lru_slot.go index 077219b75..14ee3b50f 100644 --- a/pkg/localcache/lru/lru_slot.go +++ b/pkg/localcache/lru/lru_slot.go @@ -35,7 +35,7 @@ type slotLRU[K comparable, V any] struct { func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) { var ( slotKeys = make(map[uint64][]K) - vs = make(map[K]V) + kVs = make(map[K]V) ) for _, k := range keys { @@ -49,10 +49,10 @@ func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error) return nil, err } for key, value := range batches { - vs[key] = value + kVs[key] = value } } - return vs, nil + return kVs, nil } func (x *slotLRU[K, V]) getIndex(k K) uint64 { diff --git a/pkg/msgprocessor/conversation.go b/pkg/msgprocessor/conversation.go index 04d772d16..137eb4457 100644 --- a/pkg/msgprocessor/conversation.go +++ b/pkg/msgprocessor/conversation.go @@ -109,7 +109,7 @@ func GetConversationIDBySessionType(sessionType int, ids ...string) string { case constant.ReadGroupChatType: return "sg_" + ids[0] // super group chat case constant.NotificationChatType: - return "sn_" + ids[0] // server notification chat + return "sn_" + strings.Join(ids, "_") // server notification chat } return "" } diff --git a/pkg/rpccache/auth.go b/pkg/rpccache/auth.go new file mode 100644 index 000000000..dd8c18627 --- /dev/null +++ b/pkg/rpccache/auth.go @@ -0,0 +1,69 @@ +package rpccache + +import ( + "context" + "time" + + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/open-im-server/v3/pkg/common/convert" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" + "github.com/openimsdk/open-im-server/v3/pkg/localcache" + "github.com/openimsdk/open-im-server/v3/pkg/rpcli" + "github.com/openimsdk/protocol/auth" + "github.com/openimsdk/tools/log" + "github.com/redis/go-redis/v9" +) + +func NewAuthLocalCache(client *rpcli.AuthClient, localCache *config.LocalCache, cli redis.UniversalClient) *AuthLocalCache { + lc := localCache.Auth + log.ZDebug(context.Background(), "AuthLocalCache", "topic", lc.Topic, "slotNum", lc.SlotNum, "slotSize", lc.SlotSize, "enable", lc.Enable()) + x := &AuthLocalCache{ + client: client, + local: localcache.New[[]byte]( + localcache.WithLocalSlotNum(lc.SlotNum), + localcache.WithLocalSlotSize(lc.SlotSize), + localcache.WithLinkSlotNum(lc.SlotNum), + localcache.WithLocalSuccessTTL(lc.Success()), + localcache.WithLocalFailedTTL(lc.Failed()), + ), + } + if lc.Enable() { + go subscriberRedisDeleteCache(context.Background(), cli, lc.Topic, x.local.DelLocal) + } + return x +} + +type AuthLocalCache struct { + client *rpcli.AuthClient + local localcache.Cache[[]byte] +} + +func (a *AuthLocalCache) GetExistingToken(ctx context.Context, userID string, platformID int) (val map[string]int, err error) { + resp, err := a.getExistingToken(ctx, userID, platformID) + if err != nil { + return nil, err + } + + res := convert.TokenMapPb2DB(resp.TokenStates) + + return res, nil +} + +func (a *AuthLocalCache) getExistingToken(ctx context.Context, userID string, platformID int) (val *auth.GetExistingTokenResp, err error) { + start := time.Now() + log.ZDebug(ctx, "AuthLocalCache GetExistingToken req", "userID", userID, "platformID", platformID) + defer func() { + if err != nil { + log.ZError(ctx, "AuthLocalCache GetExistingToken error", err, "cost", time.Since(start), "userID", userID, "platformID", platformID) + } else { + log.ZDebug(ctx, "AuthLocalCache GetExistingToken resp", "cost", time.Since(start), "userID", userID, "platformID", platformID, "val", val) + } + }() + + var cache cacheProto[auth.GetExistingTokenResp] + + return cache.Unmarshal(a.local.Get(ctx, cachekey.GetTokenKey(userID, platformID), func(ctx context.Context) ([]byte, error) { + log.ZDebug(ctx, "AuthLocalCache GetExistingToken call rpc", "userID", userID, "platformID", platformID) + return cache.Marshal(a.client.AuthClient.GetExistingToken(ctx, &auth.GetExistingTokenReq{UserID: userID, PlatformID: int32(platformID)})) + })) +} diff --git a/pkg/rpccache/online.go b/pkg/rpccache/online.go index 9b8c03101..fb7f18393 100644 --- a/pkg/rpccache/online.go +++ b/pkg/rpccache/online.go @@ -64,7 +64,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis. case false: log.ZDebug(ctx, "fullUserCache is false") x.lruCache = lru.NewSlotLRU(1024, localcache.LRUStringHash, func() lru.LRU[string, []int32] { - return lru.NewLayLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {}) + return lru.NewLazyLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {}) }) x.CurrentPhase.Store(DoSubscribeOver) x.Cond.Broadcast()