diff --git a/.env b/.env index 0c02bf9f6b..2b903c9491 100644 --- a/.env +++ b/.env @@ -1,6 +1,5 @@ -MONGO_IMAGE=mongo:6.0.2 +MONGO_IMAGE=mongo:7.0 REDIS_IMAGE=redis:7.0.0 -ZOOKEEPER_IMAGE=bitnami/zookeeper:3.8 KAFKA_IMAGE=bitnami/kafka:3.5.1 MINIO_IMAGE=minio/minio:RELEASE.2024-01-11T07-46-16Z ETCD_IMAGE=quay.io/coreos/etcd:v3.5.13 @@ -16,4 +15,3 @@ OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.2 #OPENIM_ADMIN_FRONT_IMAGE=registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:release-v1.8.2 DATA_DIR=./ - diff --git a/.github/workflows/changelog.yml b/.github/workflows/changelog.yml new file mode 100644 index 0000000000..b97036d91e --- /dev/null +++ b/.github/workflows/changelog.yml @@ -0,0 +1,78 @@ +name: Release Changelog + +on: + release: + types: [released] + +permissions: + contents: write + pull-requests: write + +jobs: + update-changelog: + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Run Go Changelog Generator + run: | + # Run the Go changelog generator, passing the release tag if available + if [ "${{ github.event.release.tag_name }}" = "latest" ]; then + go run tools/changelog/changelog.go > "${{ github.event.release.tag_name }}-changelog.md" + else + go run tools/changelog/changelog.go "${{ github.event.release.tag_name }}" > "${{ github.event.release.tag_name }}-changelog.md" + fi + + - name: Handle changelog files + run: | + # Ensure that the CHANGELOG directory exists + mkdir -p CHANGELOG + + # Extract Major.Minor version by removing the 'v' prefix from the tag name + TAG_NAME=${{ github.event.release.tag_name }} + CHANGELOG_VERSION_NUMBER=$(echo "$TAG_NAME" | sed 's/^v//' | grep -oP '^\d+\.\d+') + + # Define the new changelog file path + CHANGELOG_FILENAME="CHANGELOG-$CHANGELOG_VERSION_NUMBER.md" + CHANGELOG_PATH="CHANGELOG/$CHANGELOG_FILENAME" + + # Check if the changelog file for the current release already exists + if [ -f "$CHANGELOG_PATH" ]; then + # If the file exists, append the new changelog to the existing one + cat "$CHANGELOG_PATH" >> "${TAG_NAME}-changelog.md" + # Overwrite the existing changelog with the updated content + mv "${TAG_NAME}-changelog.md" "$CHANGELOG_PATH" + else + # If the changelog file doesn't exist, rename the temp changelog file to the new changelog file + mv "${TAG_NAME}-changelog.md" "$CHANGELOG_PATH" + + # Ensure that README.md exists + if [ ! -f "CHANGELOG/README.md" ]; then + echo -e "# CHANGELOGs\n\n" > CHANGELOG/README.md + fi + + # Add the new changelog entry at the top of the README.md + if ! grep -q "\[$CHANGELOG_FILENAME\]" CHANGELOG/README.md; then + sed -i "3i- [$CHANGELOG_FILENAME](./$CHANGELOG_FILENAME)" CHANGELOG/README.md + # Remove the extra newline character added by sed + # sed -i '4d' CHANGELOG/README.md + fi + fi + + - name: Clean up + run: | + # Remove any temporary files that were created during the process + rm -f "${{ github.event.release.tag_name }}-changelog.md" + + - name: Create Pull Request + uses: peter-evans/create-pull-request@v7.0.5 + with: + token: ${{ secrets.GITHUB_TOKEN }} + commit-message: "Update CHANGELOG for release ${{ github.event.release.tag_name }}" + title: "Update CHANGELOG for release ${{ github.event.release.tag_name }}" + body: "This PR updates the CHANGELOG files for release ${{ github.event.release.tag_name }}" + branch: changelog-${{ github.event.release.tag_name }} + base: main + delete-branch: true + labels: changelog diff --git a/.github/workflows/go-build-test.yml b/.github/workflows/go-build-test.yml index 5b37bf47db..2ca960cc9b 100644 --- a/.github/workflows/go-build-test.yml +++ b/.github/workflows/go-build-test.yml @@ -149,7 +149,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - go_version: ["1.21"] + go_version: ["1.22"] steps: - name: Checkout Repository diff --git a/.github/workflows/update-version-file-on-release.yml b/.github/workflows/update-version-file-on-release.yml new file mode 100644 index 0000000000..113537fd9d --- /dev/null +++ b/.github/workflows/update-version-file-on-release.yml @@ -0,0 +1,84 @@ +name: Update Version File on Release + +on: + release: + types: [created] + +jobs: + update-version: + runs-on: ubuntu-latest + env: + TAG_VERSION: ${{ github.event.release.tag_name }} + steps: + # Step 1: Checkout the original repository's code + - name: Checkout code + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + # Step 2: Set up Git with official account + - name: Set up Git + run: | + git config user.name "github-actions[bot]" + git config user.email "github-actions[bot]@users.noreply.github.com" + + # Step 3: Check and delete existing tag + - name: Check and delete existing tag + run: | + if git rev-parse ${{ env.TAG_VERSION }} >/dev/null 2>&1; then + git tag -d ${{ env.TAG_VERSION }} + git push --delete origin ${{ env.TAG_VERSION }} + fi + + # Step 4: Update version file + - name: Update version file + run: | + echo "${{ env.TAG_VERSION }}" > version/version + + # Step 5: Commit and push changes + - name: Commit and push changes + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + git add version/version + git commit -m "Update version to ${{ env.TAG_VERSION }}" + git push origin HEAD:${{ github.ref }} + + # Step 6: Create and push tag + - name: Create and push tag + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + git tag ${{ env.TAG_VERSION }} + git push origin ${{ env.TAG_VERSION }} + + # Step 7: Find and Publish Draft Release + - name: Find and Publish Draft Release + uses: actions/github-script@v6 + with: + github-token: ${{ secrets.GITHUB_TOKEN }} + script: | + // Get the list of releases + const releases = await github.rest.repos.listReleases({ + owner: context.repo.owner, + repo: context.repo.repo + }); + + // Find the draft release where the title and tag_name are the same + const draftRelease = releases.data.find(release => + release.draft && release.name === release.tag_name + ); + + if (draftRelease) { + // Publish the draft release using the release_id + await github.rest.repos.updateRelease({ + owner: context.repo.owner, + repo: context.repo.repo, + release_id: draftRelease.id, // Use release_id + draft: false + }); + + core.info(`Draft Release ${draftRelease.tag_name} published successfully.`); + } else { + core.info("No matching draft release found."); + } \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index f8cfbda9ef..4b38d711bf 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ -# Use Go 1.21 Alpine as the base image for building the application -FROM golang:1.21-alpine AS builder +# Use Go 1.22 Alpine as the base image for building the application +FROM golang:1.22-alpine AS builder # Define the base directory for the application as an environment variable ENV SERVER_DIR=/openim-server @@ -22,7 +22,7 @@ RUN go install github.com/magefile/mage@v1.15.0 RUN mage build # Using Alpine Linux with Go environment for the final image -FROM golang:1.21-alpine +FROM golang:1.22-alpine # Install necessary packages, such as bash RUN apk add --no-cache bash diff --git a/cmd/openim-api/main.go b/cmd/openim-api/main.go index e29ed2a592..3690cfc991 100644 --- a/cmd/openim-api/main.go +++ b/cmd/openim-api/main.go @@ -15,10 +15,9 @@ package main import ( - _ "net/http/pprof" - "github.com/openimsdk/open-im-server/v3/pkg/common/cmd" "github.com/openimsdk/tools/system/program" + _ "net/http/pprof" ) func main() { diff --git a/config/mongodb.yml b/config/mongodb.yml index 78f85992c9..072cb4b8f5 100644 --- a/config/mongodb.yml +++ b/config/mongodb.yml @@ -8,6 +8,8 @@ database: openim_v3 username: openIM # Password for database authentication password: openIM123 +# Authentication source for database authentication, if use root user, set it to admin +authSource: openim_v3 # Maximum number of connections in the connection pool maxPoolSize: 100 # Maximum number of retry attempts for a failed database connection diff --git a/config/openim-msggateway.yml b/config/openim-msggateway.yml index 5659c6f9b3..6c46b52a81 100644 --- a/config/openim-msggateway.yml +++ b/config/openim-msggateway.yml @@ -22,5 +22,3 @@ longConnSvr: websocketMaxMsgLen: 4096 # WebSocket connection handshake timeout in seconds websocketTimeout: 10 - - diff --git a/config/openim-push.yml b/config/openim-push.yml index 4d2aaca6b0..70e67add2b 100644 --- a/config/openim-push.yml +++ b/config/openim-push.yml @@ -13,29 +13,29 @@ prometheus: ports: [ 12170, 12171, 12172, 12173, 12174, 12175, 12176, 12177, 12178, 12179, 12180, 12182, 12183, 12184, 12185, 12186 ] maxConcurrentWorkers: 3 -#Use geTui for offline push notifications, or choose fcm or jpns; corresponding configuration settings must be specified. +#Use geTui for offline push notifications, or choose fcm or jpush; corresponding configuration settings must be specified. enable: geTui geTui: pushUrl: https://restapi.getui.com/v2/$appId - masterSecret: - appKey: - intent: - channelID: - channelName: + masterSecret: + appKey: + intent: + channelID: + channelName: fcm: # Prioritize using file paths. If the file path is empty, use URL - filePath: # File path is concatenated with the parameters passed in through - c(`mage` default pass in `config/`) and filePath. + filePath: # File path is concatenated with the parameters passed in through - c(`mage` default pass in `config/`) and filePath. authURL: # Must start with https or http. -jpns: - appKey: - masterSecret: - pushURL: - pushIntent: +jpush: + appKey: + masterSecret: + pushURL: + pushIntent: # iOS system push sound and badge count iosPush: - pushSound: xxx - badgeCount: true - production: false + pushSound: xxx + badgeCount: true + production: false fullUserCache: true diff --git a/config/share.yml b/config/share.yml index 7d977ae150..1726af2dc8 100644 --- a/config/share.yml +++ b/config/share.yml @@ -15,15 +15,4 @@ imAdminUserID: [ imAdmin ] # 1: For Android, iOS, Windows, Mac, and web platforms, only one instance can be online at a time multiLogin: policy: 1 - maxNumOneEnd: 30 - customizeLoginNum: - ios: 1 - android: 1 - windows: 1 - osx: 1 - web: 1 - miniWeb: 1 - linux: 1 - aPad: 1 - iPad: 1 - admin: 1 + maxNumOneEnd: 30 \ No newline at end of file diff --git a/deployments/templates/config.yaml b/deployments/templates/config.yaml index fee0bf90a4..c108de5e8c 100644 --- a/deployments/templates/config.yaml +++ b/deployments/templates/config.yaml @@ -240,11 +240,11 @@ push: channelName: ${GETUI_CHANNEL_NAME} fcm: serviceAccount: "${FCM_SERVICE_ACCOUNT}" - jpns: - appKey: ${JPNS_APP_KEY} - masterSecret: ${JPNS_MASTER_SECRET} - pushUrl: ${JPNS_PUSH_URL} - pushIntent: ${JPNS_PUSH_INTENT} + jpush: + appKey: ${JPUSH_APP_KEY} + masterSecret: ${JPUSH_MASTER_SECRET} + pushUrl: ${JPUSH_PUSH_URL} + pushIntent: ${JPUSH_PUSH_INTENT} # App manager configuration # diff --git a/docker-compose.yml b/docker-compose.yml index 6d88bac10e..8d25383bc5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,12 +8,35 @@ services: ports: - "37017:27017" container_name: mongo - command: ["/bin/bash", "-c", "/docker-entrypoint-initdb.d/mongo-init.sh; docker-entrypoint.sh mongod --wiredTigerCacheSizeGB 1 --auth"] + command: > + bash -c ' + docker-entrypoint.sh mongod --wiredTigerCacheSizeGB $$wiredTigerCacheSizeGB --auth & + until mongosh -u $$MONGO_INITDB_ROOT_USERNAME -p $$MONGO_INITDB_ROOT_PASSWORD --authenticationDatabase admin --eval "db.runCommand({ ping: 1 })" &>/dev/null; do + echo "Waiting for MongoDB to start..." + sleep 1 + done && + mongosh -u $$MONGO_INITDB_ROOT_USERNAME -p $$MONGO_INITDB_ROOT_PASSWORD --authenticationDatabase admin --eval " + db = db.getSiblingDB(\"$$MONGO_INITDB_DATABASE\"); + if (!db.getUser(\"$$MONGO_OPENIM_USERNAME\")) { + db.createUser({ + user: \"$$MONGO_OPENIM_USERNAME\", + pwd: \"$$MONGO_OPENIM_PASSWORD\", + roles: [{role: \"readWrite\", db: \"$$MONGO_INITDB_DATABASE\"}] + }); + print(\"User created successfully: \"); + print(\"Username: $$MONGO_OPENIM_USERNAME\"); + print(\"Password: $$MONGO_OPENIM_PASSWORD\"); + print(\"Database: $$MONGO_INITDB_DATABASE\"); + } else { + print(\"User already exists in database: $$MONGO_INITDB_DATABASE, Username: $$MONGO_OPENIM_USERNAME\"); + } + " && + tail -f /dev/null + ' volumes: - "${DATA_DIR}/components/mongodb/data/db:/data/db" - "${DATA_DIR}/components/mongodb/data/logs:/data/logs" - "${DATA_DIR}/components/mongodb/data/conf:/etc/mongo" - - "./scripts/mongo-init.sh:/docker-entrypoint-initdb.d/mongo-init.sh:ro" environment: - TZ=Asia/Shanghai - wiredTigerCacheSizeGB=1 @@ -71,10 +94,7 @@ services: ports: - "19094:9094" volumes: - - ./scripts/create-topic.sh:/opt/bitnami/kafka/create-topic.sh - "${DATA_DIR}/components/kafka:/bitnami/kafka" - command: > - bash -c "/opt/bitnami/scripts/kafka/run.sh & /opt/bitnami/kafka/create-topic.sh; wait" environment: #KAFKA_HEAP_OPTS: "-Xms128m -Xmx256m" TZ: Asia/Shanghai @@ -85,10 +105,11 @@ services: KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:19094 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_NUM_PARTITIONS: 8 + KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true" networks: - openim - minio: image: "${MINIO_IMAGE}" ports: @@ -124,7 +145,7 @@ services: - "11002:80" networks: - openim - + # prometheus: # image: ${PROMETHEUS_IMAGE} # container_name: prometheus @@ -171,4 +192,3 @@ services: # - ${DATA_DIR:-./}/components/grafana:/var/lib/grafana # networks: # - openim - diff --git a/docs/contrib/environment.md b/docs/contrib/environment.md index d2db7cbf3d..0b10abc969 100644 --- a/docs/contrib/environment.md +++ b/docs/contrib/environment.md @@ -474,10 +474,10 @@ This section involves setting up additional configuration variables for Websocke | GETUI_CHANNEL_ID | [User Defined] | GeTui Channel ID | | GETUI_CHANNEL_NAME | [User Defined] | GeTui Channel Name | | FCM_SERVICE_ACCOUNT | "x.json" | FCM Service Account | -| JPNS_APP_KEY | [User Defined] | JPNS Application Key | -| JPNS_MASTER_SECRET | [User Defined] | JPNS Master Secret | -| JPNS_PUSH_URL | [User Defined] | JPNS Push Notification URL | -| JPNS_PUSH_INTENT | [User Defined] | JPNS Push Intent | +| JPUSH_APP_KEY | [User Defined] | JPUSH Application Key | +| JPUSH_MASTER_SECRET | [User Defined] | JPUSH Master Secret | +| JPUSH_PUSH_URL | [User Defined] | JPUSH Push Notification URL | +| JPUSH_PUSH_INTENT | [User Defined] | JPUSH Push Intent | | IM_ADMIN_USERID | "imAdmin" | IM Administrator ID | | IM_ADMIN_NAME | "imAdmin" | IM Administrator Nickname | | MULTILOGIN_POLICY | "1" | Multi-login Policy | diff --git a/go.mod b/go.mod index 975abd0b4b..d39ad1d346 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,8 @@ module github.com/openimsdk/open-im-server/v3 -go 1.21.2 +go 1.22.0 + +toolchain go1.23.2 require ( firebase.google.com/go/v4 v4.14.1 @@ -8,12 +10,12 @@ require ( github.com/gin-gonic/gin v1.9.1 github.com/go-playground/validator/v10 v10.20.0 github.com/gogo/protobuf v1.3.2 // indirect - github.com/golang-jwt/jwt/v4 v4.5.0 + github.com/golang-jwt/jwt/v4 v4.5.1 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.72-pre-release-v3.8.2-alpha.5 - github.com/openimsdk/tools v0.0.50-alpha.16 + github.com/openimsdk/protocol v0.0.72-alpha.54 + github.com/openimsdk/tools v0.0.50-alpha.32 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 github.com/stretchr/testify v1.9.0 @@ -92,7 +94,7 @@ require ( github.com/fsnotify/fsnotify v1.7.0 // indirect github.com/gabriel-vasile/mimetype v1.4.3 // indirect github.com/gin-contrib/sse v0.1.0 // indirect - github.com/go-logr/logr v1.4.1 // indirect + github.com/go-logr/logr v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect diff --git a/go.sum b/go.sum index ffcac60376..df5a345163 100644 --- a/go.sum +++ b/go.sum @@ -126,8 +126,8 @@ github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg= github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ= -github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= +github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= @@ -158,8 +158,8 @@ github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang-jwt/jwt/v4 v4.4.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= -github.com/golang-jwt/jwt/v4 v4.5.0 h1:7cYmW1XlMY7h7ii7UhUyChSgS5wUJEnm9uZVTGqOWzg= -github.com/golang-jwt/jwt/v4 v4.5.0/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= +github.com/golang-jwt/jwt/v4 v4.5.1 h1:JdqV9zKUdtaa9gdPlywC3aeoEsR681PlKC+4F5gQgeo= +github.com/golang-jwt/jwt/v4 v4.5.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE= @@ -319,10 +319,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.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y= github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= -github.com/openimsdk/protocol v0.0.72-pre-release-v3.8.2-alpha.5 h1:b0JAuBhzIYirHeXp7asB04bE1q+KhU3dpAaAroc/Am0= -github.com/openimsdk/protocol v0.0.72-pre-release-v3.8.2-alpha.5/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= -github.com/openimsdk/tools v0.0.50-alpha.16 h1:bC1AQvJMuOHtZm8LZRvN8L5mH1Ws2VYdL+TLTs1iGSc= -github.com/openimsdk/tools v0.0.50-alpha.16/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4= +github.com/openimsdk/protocol v0.0.72-alpha.54 h1:opato7N4QjjRq/SHD54bDSVBpOEEDp1VLWVk5Os2A9s= +github.com/openimsdk/protocol v0.0.72-alpha.54/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= +github.com/openimsdk/tools v0.0.50-alpha.32 h1:JEsUFHFnaYg230TG+Ke3SUnaA2h44t4kABAzEdv5VZw= +github.com/openimsdk/tools v0.0.50-alpha.32/go.mod h1:r5U6RbxcR4xhKb2fhTmKGC9Yt5LcErHBVt3lhXQIHSo= 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= @@ -356,8 +356,8 @@ github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE= -github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M= -github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc= github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= diff --git a/internal/api/jssdk/jssdk.go b/internal/api/jssdk/jssdk.go index 7f136c74ce..2fcbf5ec66 100644 --- a/internal/api/jssdk/jssdk.go +++ b/internal/api/jssdk/jssdk.go @@ -1,11 +1,15 @@ package jssdk import ( + "context" "github.com/gin-gonic/gin" "github.com/openimsdk/protocol/conversation" + "github.com/openimsdk/protocol/group" + "github.com/openimsdk/protocol/jssdk" "github.com/openimsdk/protocol/msg" + "github.com/openimsdk/protocol/relation" "github.com/openimsdk/protocol/sdkws" - "github.com/openimsdk/tools/a2r" + "github.com/openimsdk/protocol/user" "github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/utils/datautil" "sort" @@ -16,16 +20,22 @@ const ( defaultGetActiveConversation = 100 ) -func NewJSSdkApi(msg msg.MsgClient, conv conversation.ConversationClient) *JSSdk { +func NewJSSdkApi(user user.UserClient, friend relation.FriendClient, group group.GroupClient, msg msg.MsgClient, conv conversation.ConversationClient) *JSSdk { return &JSSdk{ - msg: msg, - conv: conv, + user: user, + friend: friend, + group: group, + msg: msg, + conv: conv, } } type JSSdk struct { - msg msg.MsgClient - conv conversation.ConversationClient + user user.UserClient + friend relation.FriendClient + group group.GroupClient + msg msg.MsgClient + conv conversation.ConversationClient } func (x *JSSdk) GetActiveConversations(c *gin.Context) { @@ -36,25 +46,71 @@ func (x *JSSdk) GetConversations(c *gin.Context) { call(c, x.getConversations) } -func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, error) { - req, err := a2r.ParseRequest[ActiveConversationsReq](ctx) - if err != nil { - return nil, err +func (x *JSSdk) fillConversations(ctx context.Context, conversations []*jssdk.ConversationMsg) error { + if len(conversations) == 0 { + return nil + } + var ( + userIDs []string + groupIDs []string + ) + for _, c := range conversations { + if c.Conversation.GroupID == "" { + userIDs = append(userIDs, c.Conversation.UserID) + } else { + groupIDs = append(groupIDs, c.Conversation.GroupID) + } + } + var ( + userMap map[string]*sdkws.UserInfo + friendMap map[string]*relation.FriendInfoOnly + groupMap map[string]*sdkws.GroupInfo + ) + if len(userIDs) > 0 { + users, err := field(ctx, x.user.GetDesignateUsers, &user.GetDesignateUsersReq{UserIDs: userIDs}, (*user.GetDesignateUsersResp).GetUsersInfo) + if err != nil { + return err + } + friends, err := field(ctx, x.friend.GetFriendInfo, &relation.GetFriendInfoReq{OwnerUserID: conversations[0].Conversation.OwnerUserID, FriendUserIDs: userIDs}, (*relation.GetFriendInfoResp).GetFriendInfos) + if err != nil { + return err + } + userMap = datautil.SliceToMap(users, (*sdkws.UserInfo).GetUserID) + friendMap = datautil.SliceToMap(friends, (*relation.FriendInfoOnly).GetFriendUserID) + } + if len(groupIDs) > 0 { + resp, err := x.group.GetGroupsInfo(ctx, &group.GetGroupsInfoReq{GroupIDs: groupIDs}) + if err != nil { + return err + } + groupMap = datautil.SliceToMap(resp.GroupInfos, (*sdkws.GroupInfo).GetGroupID) } + for _, c := range conversations { + if c.Conversation.GroupID == "" { + c.User = userMap[c.Conversation.UserID] + c.Friend = friendMap[c.Conversation.UserID] + } else { + c.Group = groupMap[c.Conversation.GroupID] + } + } + return nil +} + +func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActiveConversationsReq) (*jssdk.GetActiveConversationsResp, error) { if req.Count <= 0 || req.Count > maxGetActiveConversation { req.Count = defaultGetActiveConversation } - opUserID := mcontext.GetOpUserID(ctx) + req.OwnerUserID = mcontext.GetOpUserID(ctx) conversationIDs, err := field(ctx, x.conv.GetConversationIDs, - &conversation.GetConversationIDsReq{UserID: opUserID}, (*conversation.GetConversationIDsResp).GetConversationIDs) + &conversation.GetConversationIDsReq{UserID: req.OwnerUserID}, (*conversation.GetConversationIDsResp).GetConversationIDs) if err != nil { return nil, err } if len(conversationIDs) == 0 { - return &ConversationsResp{}, nil + return &jssdk.GetActiveConversationsResp{}, nil } readSeq, err := field(ctx, x.msg.GetHasReadSeqs, - &msg.GetHasReadSeqsReq{UserID: opUserID, ConversationIDs: conversationIDs}, (*msg.SeqsInfoResp).GetMaxSeqs) + &msg.GetHasReadSeqsReq{UserID: req.OwnerUserID, ConversationIDs: conversationIDs}, (*msg.SeqsInfoResp).GetMaxSeqs) if err != nil { return nil, err } @@ -64,24 +120,24 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er return nil, err } if len(activeConversation) == 0 { - return &ConversationsResp{}, nil + return &jssdk.GetActiveConversationsResp{}, nil } sortConversations := sortActiveConversations{ Conversation: activeConversation, } if len(activeConversation) > 1 { pinnedConversationIDs, err := field(ctx, x.conv.GetPinnedConversationIDs, - &conversation.GetPinnedConversationIDsReq{UserID: opUserID}, (*conversation.GetPinnedConversationIDsResp).GetConversationIDs) + &conversation.GetPinnedConversationIDsReq{UserID: req.OwnerUserID}, (*conversation.GetPinnedConversationIDsResp).GetConversationIDs) if err != nil { return nil, err } sortConversations.PinnedConversationIDs = datautil.SliceSet(pinnedConversationIDs) } sort.Sort(&sortConversations) - sortList := sortConversations.Top(req.Count) + sortList := sortConversations.Top(int(req.Count)) conversations, err := field(ctx, x.conv.GetConversations, &conversation.GetConversationsReq{ - OwnerUserID: opUserID, + OwnerUserID: req.OwnerUserID, ConversationIDs: datautil.Slice(sortList, func(c *msg.ActiveConversation) string { return c.ConversationID })}, (*conversation.GetConversationsResp).GetConversations) @@ -90,7 +146,7 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er } msgs, err := field(ctx, x.msg.GetSeqMessage, &msg.GetSeqMessageReq{ - UserID: opUserID, + UserID: req.OwnerUserID, Conversations: datautil.Slice(sortList, func(c *msg.ActiveConversation) *msg.ConversationSeqs { return &msg.ConversationSeqs{ ConversationID: c.ConversationID, @@ -104,7 +160,7 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er conversationMap := datautil.SliceToMap(conversations, func(c *conversation.Conversation) string { return c.ConversationID }) - resp := make([]ConversationMsg, 0, len(sortList)) + resp := make([]*jssdk.ConversationMsg, 0, len(sortList)) for _, c := range sortList { conv, ok := conversationMap[c.ConversationID] if !ok { @@ -114,13 +170,16 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 { lastMsg = msgList.Msgs[0] } - resp = append(resp, ConversationMsg{ + resp = append(resp, &jssdk.ConversationMsg{ Conversation: conv, LastMsg: lastMsg, MaxSeq: c.MaxSeq, ReadSeq: readSeq[c.ConversationID], }) } + if err := x.fillConversations(ctx, resp); err != nil { + return nil, err + } var unreadCount int64 for _, c := range activeConversation { count := c.MaxSeq - readSeq[c.ConversationID] @@ -128,24 +187,20 @@ func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, er unreadCount += count } } - return &ConversationsResp{ + return &jssdk.GetActiveConversationsResp{ Conversations: resp, UnreadCount: unreadCount, }, nil } -func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) { - req, err := a2r.ParseRequest[conversation.GetConversationsReq](ctx) - if err != nil { - return nil, err - } +func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversationsReq) (*jssdk.GetConversationsResp, error) { req.OwnerUserID = mcontext.GetOpUserID(ctx) - conversations, err := field(ctx, x.conv.GetConversations, req, (*conversation.GetConversationsResp).GetConversations) + conversations, err := field(ctx, x.conv.GetConversations, &conversation.GetConversationsReq{OwnerUserID: req.OwnerUserID, ConversationIDs: req.ConversationIDs}, (*conversation.GetConversationsResp).GetConversations) if err != nil { return nil, err } if len(conversations) == 0 { - return &ConversationsResp{}, nil + return &jssdk.GetConversationsResp{}, nil } req.ConversationIDs = datautil.Slice(conversations, func(c *conversation.Conversation) string { return c.ConversationID @@ -177,19 +232,22 @@ func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) { return nil, err } } - resp := make([]ConversationMsg, 0, len(conversations)) + resp := make([]*jssdk.ConversationMsg, 0, len(conversations)) for _, c := range conversations { var lastMsg *sdkws.MsgData if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 { lastMsg = msgList.Msgs[0] } - resp = append(resp, ConversationMsg{ + resp = append(resp, &jssdk.ConversationMsg{ Conversation: c, LastMsg: lastMsg, MaxSeq: maxSeqs[c.ConversationID], ReadSeq: readSeqs[c.ConversationID], }) } + if err := x.fillConversations(ctx, resp); err != nil { + return nil, err + } var unreadCount int64 for conversationID, maxSeq := range maxSeqs { count := maxSeq - readSeqs[conversationID] @@ -197,7 +255,7 @@ func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) { unreadCount += count } } - return &ConversationsResp{ + return &jssdk.GetConversationsResp{ Conversations: resp, UnreadCount: unreadCount, }, nil diff --git a/internal/api/jssdk/stu.go b/internal/api/jssdk/stu.go deleted file mode 100644 index 2f63975b3b..0000000000 --- a/internal/api/jssdk/stu.go +++ /dev/null @@ -1,22 +0,0 @@ -package jssdk - -import ( - "github.com/openimsdk/protocol/conversation" - "github.com/openimsdk/protocol/sdkws" -) - -type ActiveConversationsReq struct { - Count int `json:"count"` -} - -type ConversationMsg struct { - Conversation *conversation.Conversation `json:"conversation"` - LastMsg *sdkws.MsgData `json:"lastMsg"` - MaxSeq int64 `json:"maxSeq"` - ReadSeq int64 `json:"readSeq"` -} - -type ConversationsResp struct { - UnreadCount int64 `json:"unreadCount"` - Conversations []ConversationMsg `json:"conversations"` -} diff --git a/internal/api/jssdk/tools.go b/internal/api/jssdk/tools.go index c57457d9f4..c19d8970b6 100644 --- a/internal/api/jssdk/tools.go +++ b/internal/api/jssdk/tools.go @@ -3,8 +3,14 @@ package jssdk import ( "context" "github.com/gin-gonic/gin" + "github.com/openimsdk/tools/a2r" "github.com/openimsdk/tools/apiresp" + "github.com/openimsdk/tools/checker" + "github.com/openimsdk/tools/errs" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" + "io" + "strings" ) func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A, opts ...grpc.CallOption) (*B, error), req *A, get func(*B) C) (C, error) { @@ -16,11 +22,56 @@ func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A return get(resp), nil } -func call[R any](c *gin.Context, fn func(ctx *gin.Context) (R, error)) { - resp, err := fn(c) +func call[A, B any](c *gin.Context, fn func(ctx context.Context, req *A) (*B, error)) { + var isJSON bool + switch contentType := c.GetHeader("Content-Type"); { + case contentType == "": + isJSON = true + case strings.Contains(contentType, "application/json"): + isJSON = true + case strings.Contains(contentType, "application/protobuf"): + case strings.Contains(contentType, "application/x-protobuf"): + default: + apiresp.GinError(c, errs.ErrArgs.WrapMsg("unsupported content type")) + return + } + var req *A + if isJSON { + var err error + req, err = a2r.ParseRequest[A](c) + if err != nil { + apiresp.GinError(c, err) + return + } + } else { + body, err := io.ReadAll(c.Request.Body) + if err != nil { + apiresp.GinError(c, err) + return + } + req = new(A) + if err := proto.Unmarshal(body, any(req).(proto.Message)); err != nil { + apiresp.GinError(c, err) + return + } + if err := checker.Validate(&req); err != nil { + apiresp.GinError(c, err) + return + } + } + resp, err := fn(c, req) + if err != nil { + apiresp.GinError(c, err) + return + } + if isJSON { + apiresp.GinSuccess(c, resp) + return + } + body, err := proto.Marshal(any(resp).(proto.Message)) if err != nil { apiresp.GinError(c, err) return } - apiresp.GinSuccess(c, resp) + apiresp.GinSuccess(c, body) } diff --git a/internal/api/jssdk_test.go b/internal/api/jssdk_test.go deleted file mode 100644 index 472ca56b57..0000000000 --- a/internal/api/jssdk_test.go +++ /dev/null @@ -1,37 +0,0 @@ -package api - -import ( - "github.com/openimsdk/protocol/msg" - "sort" - "testing" -) - -func TestName(t *testing.T) { - val := sortActiveConversations{ - Conversation: []*msg.ActiveConversation{ - { - ConversationID: "100", - LastTime: 100, - }, - { - ConversationID: "200", - LastTime: 200, - }, - { - ConversationID: "300", - LastTime: 300, - }, - { - ConversationID: "400", - LastTime: 400, - }, - }, - //PinnedConversationIDs: map[string]struct{}{ - // "100": {}, - // "300": {}, - //}, - } - sort.Sort(&val) - t.Log(val) - -} diff --git a/internal/api/router.go b/internal/api/router.go index f87ec526c7..85c48b2847 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -74,10 +74,10 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En case BestSpeed: r.Use(gzip.Gzip(gzip.BestSpeed)) } - r.Use(prommetricsGin(), gin.Recovery(), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc)) + r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc)) u := NewUserApi(*userRpc) m := NewMessageApi(messageRpc, userRpc, config.Share.IMAdminUserID) - j := jssdk.NewJSSdkApi(messageRpc.Client, conversationRpc.Client) + j := jssdk.NewJSSdkApi(userRpc.Client, friendRpc.Client, groupRpc.Client, messageRpc.Client, conversationRpc.Client) userRouterGroup := r.Group("/user") { userRouterGroup.POST("/user_register", u.UserRegister) diff --git a/internal/msggateway/client.go b/internal/msggateway/client.go index bc06fa9507..0da7d72208 100644 --- a/internal/msggateway/client.go +++ b/internal/msggateway/client.go @@ -16,7 +16,9 @@ package msggateway import ( "context" + "encoding/json" "fmt" + "github.com/openimsdk/tools/mw" "runtime/debug" "sync" "sync/atomic" @@ -69,6 +71,8 @@ type Client struct { IsCompress bool `json:"isCompress"` UserID string `json:"userID"` IsBackground bool `json:"isBackground"` + SDKType string `json:"sdkType"` + Encoder Encoder ctx *UserConnContext longConnServer LongConnServer closed atomic.Bool @@ -94,11 +98,17 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer c.closed.Store(false) c.closedErr = nil c.token = ctx.GetToken() + c.SDKType = ctx.GetSDKType() c.hbCtx, c.hbCancel = context.WithCancel(c.ctx) c.subLock = new(sync.Mutex) if c.subUserIDs != nil { clear(c.subUserIDs) } + if c.SDKType == GoSDK { + c.Encoder = NewGobEncoder() + } else { + c.Encoder = NewJsonEncoder() + } c.subUserIDs = make(map[string]struct{}) } @@ -159,9 +169,12 @@ func (c *Client) readMessage() { return } case MessageText: - c.closedErr = ErrNotSupportMessageProtocol - return - + _ = c.conn.SetReadDeadline(pongWait) + parseDataErr := c.handlerTextMessage(message) + if parseDataErr != nil { + c.closedErr = parseDataErr + return + } case PingMessage: err := c.writePongMsg("") log.ZError(c.ctx, "writePongMsg", err) @@ -188,7 +201,7 @@ func (c *Client) handleMessage(message []byte) error { var binaryReq = getReq() defer freeReq(binaryReq) - err := c.longConnServer.Decode(message, binaryReq) + err := c.Encoder.Decode(message, binaryReq) if err != nil { return err } @@ -335,7 +348,7 @@ func (c *Client) writeBinaryMsg(resp Resp) error { return nil } - encodedBuf, err := c.longConnServer.Encode(resp) + encodedBuf, err := c.Encoder.Encode(resp) if err != nil { return err } @@ -363,6 +376,11 @@ func (c *Client) writeBinaryMsg(resp Resp) error { func (c *Client) activeHeartbeat(ctx context.Context) { if c.PlatformID == constant.WebPlatformID { go func() { + defer func() { + if r := recover(); r != nil { + mw.PanicStackToLog(ctx, r) + } + }() log.ZDebug(ctx, "server initiative send heartbeat start.") ticker := time.NewTicker(pingPeriod) defer ticker.Stop() @@ -419,3 +437,28 @@ func (c *Client) writePongMsg(appData string) error { return errs.Wrap(err) } + +func (c *Client) handlerTextMessage(b []byte) error { + var msg TextMessage + if err := json.Unmarshal(b, &msg); err != nil { + return err + } + switch msg.Type { + case TextPong: + return nil + case TextPing: + msg.Type = TextPong + msgData, err := json.Marshal(msg) + if err != nil { + return err + } + c.w.Lock() + defer c.w.Unlock() + if err := c.conn.SetWriteDeadline(writeWait); err != nil { + return err + } + return c.conn.WriteMessage(MessageText, msgData) + default: + return fmt.Errorf("not support message type %s", msg.Type) + } +} diff --git a/internal/msggateway/constant.go b/internal/msggateway/constant.go index 584cebe1e1..a825c05196 100644 --- a/internal/msggateway/constant.go +++ b/internal/msggateway/constant.go @@ -27,6 +27,12 @@ const ( GzipCompressionProtocol = "gzip" BackgroundStatus = "isBackground" SendResponse = "isMsgResp" + SDKType = "sdkType" +) + +const ( + GoSDK = "go" + JsSDK = "js" ) const ( diff --git a/internal/msggateway/context.go b/internal/msggateway/context.go index 3909766b1b..d73a96df4c 100644 --- a/internal/msggateway/context.go +++ b/internal/msggateway/context.go @@ -153,6 +153,14 @@ func (c *UserConnContext) GetCompression() bool { return false } +func (c *UserConnContext) GetSDKType() string { + sdkType := c.Req.URL.Query().Get(SDKType) + if sdkType == "" { + sdkType = GoSDK + } + return sdkType +} + func (c *UserConnContext) ShouldSendResp() bool { errResp, exists := c.Query(SendResponse) if exists { @@ -193,7 +201,11 @@ func (c *UserConnContext) ParseEssentialArgs() error { _, err := strconv.Atoi(platformIDStr) if err != nil { return servererrs.ErrConnArgsErr.WrapMsg("platformID is not int") - + } + switch sdkType, _ := c.Query(SDKType); sdkType { + case "", GoSDK, JsSDK: + default: + return servererrs.ErrConnArgsErr.WrapMsg("sdkType is not go or js") } return nil } diff --git a/internal/msggateway/encoder.go b/internal/msggateway/encoder.go index 3af2663748..6a5936d6d2 100644 --- a/internal/msggateway/encoder.go +++ b/internal/msggateway/encoder.go @@ -17,6 +17,7 @@ package msggateway import ( "bytes" "encoding/gob" + "encoding/json" "github.com/openimsdk/tools/errs" ) @@ -28,12 +29,12 @@ type Encoder interface { type GobEncoder struct{} -func NewGobEncoder() *GobEncoder { - return &GobEncoder{} +func NewGobEncoder() Encoder { + return GobEncoder{} } -func (g *GobEncoder) Encode(data any) ([]byte, error) { - buff := bytes.Buffer{} +func (g GobEncoder) Encode(data any) ([]byte, error) { + var buff bytes.Buffer enc := gob.NewEncoder(&buff) if err := enc.Encode(data); err != nil { return nil, errs.WrapMsg(err, "GobEncoder.Encode failed", "action", "encode") @@ -41,7 +42,7 @@ func (g *GobEncoder) Encode(data any) ([]byte, error) { return buff.Bytes(), nil } -func (g *GobEncoder) Decode(encodeData []byte, decodeData any) error { +func (g GobEncoder) Decode(encodeData []byte, decodeData any) error { buff := bytes.NewBuffer(encodeData) dec := gob.NewDecoder(buff) if err := dec.Decode(decodeData); err != nil { @@ -49,3 +50,25 @@ func (g *GobEncoder) Decode(encodeData []byte, decodeData any) error { } return nil } + +type JsonEncoder struct{} + +func NewJsonEncoder() Encoder { + return JsonEncoder{} +} + +func (g JsonEncoder) Encode(data any) ([]byte, error) { + b, err := json.Marshal(data) + if err != nil { + return nil, errs.New("JsonEncoder.Encode failed", "action", "encode") + } + return b, nil +} + +func (g JsonEncoder) Decode(encodeData []byte, decodeData any) error { + err := json.Unmarshal(encodeData, decodeData) + if err != nil { + return errs.New("JsonEncoder.Decode failed", "action", "decode") + } + return nil +} diff --git a/internal/msggateway/hub_server.go b/internal/msggateway/hub_server.go index e96ab4b0dc..23d9150133 100644 --- a/internal/msggateway/hub_server.go +++ b/internal/msggateway/hub_server.go @@ -83,17 +83,11 @@ func NewServer(rpcPort int, longConnServer LongConnServer, conf *Config, ready f return s } -func (s *Server) OnlinePushMsg( - context context.Context, - req *msggateway.OnlinePushMsgReq, -) (*msggateway.OnlinePushMsgResp, error) { +func (s *Server) OnlinePushMsg(context context.Context, req *msggateway.OnlinePushMsgReq) (*msggateway.OnlinePushMsgResp, error) { panic("implement me") } -func (s *Server) GetUsersOnlineStatus( - ctx context.Context, - req *msggateway.GetUsersOnlineStatusReq, -) (*msggateway.GetUsersOnlineStatusResp, error) { +func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUsersOnlineStatusReq) (*msggateway.GetUsersOnlineStatusResp, error) { if !authverify.IsAppManagerUid(ctx, s.config.Share.IMAdminUserID) { return nil, errs.ErrNoPermission.WrapMsg("only app manager") } @@ -155,6 +149,7 @@ func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.M (client.IsBackground && client.PlatformID != constant.IOSPlatformID) { err := client.PushMessage(ctx, msgData) if err != nil { + log.ZWarn(ctx, "online push msg failed", err, "userID", userID, "platformID", client.PlatformID) userPlatform.ResultCode = int64(servererrs.ErrPushMsgErr.Code()) } else { if _, ok := s.pushTerminal[client.PlatformID]; ok { @@ -220,10 +215,7 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msgga } } -func (s *Server) KickUserOffline( - ctx context.Context, - req *msggateway.KickUserOfflineReq, -) (*msggateway.KickUserOfflineResp, error) { +func (s *Server) KickUserOffline(ctx context.Context, req *msggateway.KickUserOfflineReq) (*msggateway.KickUserOfflineResp, error) { for _, v := range req.KickUserIDList { clients, _, ok := s.LongConnServer.GetUserPlatformCons(v, int(req.PlatformID)) if !ok { diff --git a/internal/msggateway/message_handler.go b/internal/msggateway/message_handler.go index 4b78c10048..5407ba90cb 100644 --- a/internal/msggateway/message_handler.go +++ b/internal/msggateway/message_handler.go @@ -16,6 +16,7 @@ package msggateway import ( "context" + "encoding/json" "sync" "github.com/go-playground/validator/v10" @@ -31,6 +32,16 @@ import ( "github.com/openimsdk/tools/utils/jsonutil" ) +const ( + TextPing = "ping" + TextPong = "pong" +) + +type TextMessage struct { + Type string `json:"type"` + Body json.RawMessage `json:"body"` +} + type Req struct { ReqIdentifier int32 `json:"reqIdentifier" validate:"required"` Token string `json:"token"` diff --git a/internal/msggateway/ws_server.go b/internal/msggateway/ws_server.go index b92d7eb442..e6b4f3fa47 100644 --- a/internal/msggateway/ws_server.go +++ b/internal/msggateway/ws_server.go @@ -37,7 +37,6 @@ type LongConnServer interface { SetKickHandlerInfo(i *kickHandler) SubUserOnlineStatus(ctx context.Context, client *Client, data *Req) ([]byte, error) Compressor - Encoder MessageHandler } @@ -61,7 +60,7 @@ type WsServer struct { authClient *rpcclient.Auth disCov discovery.SvcDiscoveryRegistry Compressor - Encoder + //Encoder MessageHandler webhookClient *webhook.Client } @@ -135,7 +134,6 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer { clients: newUserMap(), subscription: newSubscription(), Compressor: NewGzipCompressor(), - Encoder: NewGobEncoder(), webhookClient: webhook.NewWebhookClient(msgGatewayConfig.WebhooksConfig.URL), } } @@ -278,14 +276,7 @@ func (ws *WsServer) registerClient(client *Client) { wg.Wait() - log.ZDebug( - client.ctx, - "user online", - "online user Num", - ws.onlineUserNum.Load(), - "online user conn Num", - ws.onlineUserConnNum.Load(), - ) + log.ZDebug(client.ctx, "user online", "online user Num", ws.onlineUserNum.Load(), "online user conn Num", ws.onlineUserConnNum.Load()) } func getRemoteAdders(client []*Client) string { @@ -327,11 +318,6 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien switch ws.msgGatewayConfig.Share.MultiLogin.Policy { case constant.DefalutNotKick: - case constant.WebAndOther: - if constant.PlatformIDToClass(newClient.PlatformID) == constant.WebPlatformStr { - return - } - fallthrough case constant.PCAndOther: if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC { return @@ -356,7 +342,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien log.ZWarn(newClient.ctx, "InvalidateToken err", err, "userID", newClient.UserID, "platformID", newClient.PlatformID) } - case constant.PcMobileAndWeb: + case constant.AllLoginButSameClassKick: clients, ok := ws.clients.GetAll(newClient.UserID) if !ok { return @@ -370,21 +356,6 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien } } kickTokenFunc(kickClients) - - case constant.SingleTerminalLogin: - clients, ok := ws.clients.GetAll(newClient.UserID) - if !ok { - return - } - var ( - kickClients []*Client - ) - for _, client := range clients { - kickClients = append(kickClients, client) - } - kickTokenFunc(kickClients) - case constant.Customize: - // todo } } diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go index f11cfde1af..92053931ce 100644 --- a/internal/msgtransfer/init.go +++ b/internal/msgtransfer/init.go @@ -136,6 +136,11 @@ func (m *MsgTransfer) Start(index int, config *Config) error { if config.MsgTransfer.Prometheus.Enable { go func() { + defer func() { + if r := recover(); r != nil { + mw.PanicStackToLog(m.ctx, r) + } + }() prometheusPort, err := datautil.GetElemByIndex(config.MsgTransfer.Prometheus.Ports, index) if err != nil { netErr = err diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go index 84453c8df4..4a5d5ba89e 100644 --- a/internal/msgtransfer/online_history_msg_handler.go +++ b/internal/msgtransfer/online_history_msg_handler.go @@ -19,6 +19,7 @@ import ( "encoding/json" "errors" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" + "github.com/openimsdk/tools/mw" "strconv" "strings" "sync" @@ -346,6 +347,12 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con } } func (och *OnlineHistoryRedisConsumerHandler) HandleUserHasReadSeqMessages(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + mw.PanicStackToLog(ctx, r) + } + }() + defer och.wg.Done() for msg := range och.conversationUserHasReadChan { diff --git a/internal/push/offlinepush/dummy/push.go b/internal/push/offlinepush/dummy/push.go index 09831cabfa..0bccaf4a40 100644 --- a/internal/push/offlinepush/dummy/push.go +++ b/internal/push/offlinepush/dummy/push.go @@ -29,5 +29,6 @@ type Dummy struct { func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error { log.ZDebug(ctx, "dummy push") + log.ZWarn(ctx, "Dummy push", nil, "ps", "The offline push is not configured. To configure it, please go to config/openim-push.yml.") return nil } diff --git a/internal/push/offlinepush/jpush/body/notification.go b/internal/push/offlinepush/jpush/body/notification.go index 42e59c46cf..383b3fb263 100644 --- a/internal/push/offlinepush/jpush/body/notification.go +++ b/internal/push/offlinepush/jpush/body/notification.go @@ -15,6 +15,7 @@ package body import ( + "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/pkg/common/config" ) @@ -26,38 +27,44 @@ type Notification struct { type Android struct { Alert string `json:"alert,omitempty"` + Title string `json:"title,omitempty"` Intent struct { URL string `json:"url,omitempty"` } `json:"intent,omitempty"` - Extras Extras `json:"extras"` + Extras map[string]string `json:"extras,omitempty"` } type Ios struct { - Alert string `json:"alert,omitempty"` - Sound string `json:"sound,omitempty"` - Badge string `json:"badge,omitempty"` - Extras Extras `json:"extras"` - MutableContent bool `json:"mutable-content"` + Alert IosAlert `json:"alert,omitempty"` + Sound string `json:"sound,omitempty"` + Badge string `json:"badge,omitempty"` + Extras map[string]string `json:"extras,omitempty"` + MutableContent bool `json:"mutable-content"` } -type Extras struct { - ClientMsgID string `json:"clientMsgID"` +type IosAlert struct { + Title string `json:"title,omitempty"` + Body string `json:"body,omitempty"` } -func (n *Notification) SetAlert(alert string) { +func (n *Notification) SetAlert(alert string, title string, opts *options.Opts) { n.Alert = alert n.Android.Alert = alert - n.IOS.Alert = alert - n.IOS.Sound = "default" - n.IOS.Badge = "+1" + n.Android.Title = title + n.IOS.Alert.Body = alert + n.IOS.Alert.Title = title + n.IOS.Sound = opts.IOSPushSound + if opts.IOSBadgeCount { + n.IOS.Badge = "+1" + } } -func (n *Notification) SetExtras(extras Extras) { +func (n *Notification) SetExtras(extras map[string]string) { n.IOS.Extras = extras n.Android.Extras = extras } func (n *Notification) SetAndroidIntent(pushConf *config.Push) { - n.Android.Intent.URL = pushConf.JPNS.PushIntent + n.Android.Intent.URL = pushConf.JPush.PushIntent } func (n *Notification) IOSEnableMutableContent() { diff --git a/internal/push/offlinepush/jpush/push.go b/internal/push/offlinepush/jpush/push.go index dac52597f5..2694902f20 100644 --- a/internal/push/offlinepush/jpush/push.go +++ b/internal/push/offlinepush/jpush/push.go @@ -18,9 +18,9 @@ import ( "context" "encoding/base64" "fmt" - "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush/body" + "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/tools/utils/httputil" ) @@ -57,17 +57,23 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin var au body.Audience au.SetAlias(userIDs) var no body.Notification - var extras body.Extras + extras := make(map[string]string) + extras["ex"] = opts.Ex if opts.Signal.ClientMsgID != "" { - extras.ClientMsgID = opts.Signal.ClientMsgID + extras["ClientMsgID"] = opts.Signal.ClientMsgID } no.IOSEnableMutableContent() no.SetExtras(extras) - no.SetAlert(title) + no.SetAlert(content, title, opts) no.SetAndroidIntent(j.pushConf) var msg body.Message msg.SetMsgContent(content) + msg.SetTitle(title) + if opts.Signal.ClientMsgID != "" { + msg.SetExtras("ClientMsgID", opts.Signal.ClientMsgID) + } + msg.SetExtras("ex", opts.Ex) var opt body.Options opt.SetApnsProduction(j.pushConf.IOSPush.Production) var pushObj body.PushObj @@ -76,19 +82,26 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin pushObj.SetNotification(&no) pushObj.SetMessage(&msg) pushObj.SetOptions(&opt) - var resp any - return j.request(ctx, pushObj, resp, 5) + var resp map[string]any + return j.request(ctx, pushObj, &resp, 5) } -func (j *JPush) request(ctx context.Context, po body.PushObj, resp any, timeout int) error { - return j.httpClient.PostReturn( +func (j *JPush) request(ctx context.Context, po body.PushObj, resp *map[string]any, timeout int) error { + err := j.httpClient.PostReturn( ctx, - j.pushConf.JPNS.PushURL, + j.pushConf.JPush.PushURL, map[string]string{ - "Authorization": j.getAuthorization(j.pushConf.JPNS.AppKey, j.pushConf.JPNS.MasterSecret), + "Authorization": j.getAuthorization(j.pushConf.JPush.AppKey, j.pushConf.JPush.MasterSecret), }, po, resp, timeout, ) + if err != nil { + return err + } + if (*resp)["sendno"] != "0" { + return fmt.Errorf("jpush push failed %v", resp) + } + return nil } diff --git a/internal/push/offlinepush/offlinepusher.go b/internal/push/offlinepush/offlinepusher.go index d655a924a2..17d5d7071b 100644 --- a/internal/push/offlinepush/offlinepusher.go +++ b/internal/push/offlinepush/offlinepusher.go @@ -23,8 +23,6 @@ import ( "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mcontext" "strings" ) @@ -51,7 +49,6 @@ func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPa offlinePusher = jpush.NewClient(pushConf) default: offlinePusher = dummy.NewClient() - log.ZWarn(mcontext.WithMustInfoCtx([]string{"push start", "admin", "admin", ""}), "Unknown push config", nil) } return offlinePusher, nil } diff --git a/internal/push/offlinepush_handler.go b/internal/push/offlinepush_handler.go index a80c147f4a..5c69da0054 100644 --- a/internal/push/offlinepush_handler.go +++ b/internal/push/offlinepush_handler.go @@ -73,7 +73,7 @@ func (o *OfflinePushConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (ti IsAtSelf bool `json:"isAtSelf"` } - opts = &options.Opts{Signal: &options.Signal{}} + opts = &options.Opts{Signal: &options.Signal{ClientMsgID: msg.ClientMsgID}} if msg.OfflinePushInfo != nil { opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index 41ad5962aa..b5bbed3ed5 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -4,6 +4,10 @@ import ( "context" "encoding/json" + "math/rand" + "strconv" + "time" + "github.com/IBM/sarama" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" @@ -27,9 +31,6 @@ import ( "github.com/openimsdk/tools/utils/timeutil" "github.com/redis/go-redis/v9" "google.golang.org/protobuf/proto" - "math/rand" - "strconv" - "time" ) type ConsumerHandler struct { @@ -165,17 +166,21 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg * return nil } } - offlinePushUserID := []string{msg.RecvID} + needOfflinePushUserID := []string{msg.RecvID} + var offlinePushUserID []string //receiver offline push - if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, - offlinePushUserID, msg, nil); err != nil { + if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserID, msg, &offlinePushUserID); err != nil { return err } log.ZInfo(ctx, "webhookBeforeOfflinePush end") - err = c.offlinePushMsg(ctx, msg, offlinePushUserID) + + if len(offlinePushUserID) > 0 { + needOfflinePushUserID = offlinePushUserID + } + err = c.offlinePushMsg(ctx, msg, needOfflinePushUserID) if err != nil { - log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePushUserID", offlinePushUserID, "msg", msg) + log.ZWarn(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID", needOfflinePushUserID, "msg", msg) return nil } @@ -335,6 +340,7 @@ func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID stri func (c *ConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error { title, content, opts, err := c.getOfflinePushInfos(msg) if err != nil { + log.ZError(ctx, "getOfflinePushInfos failed", err, "msg", msg) return err } err = c.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts) @@ -364,7 +370,7 @@ func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, conten IsAtSelf bool `json:"isAtSelf"` } - opts = &options.Opts{Signal: &options.Signal{}} + opts = &options.Opts{Signal: &options.Signal{ClientMsgID: msg.ClientMsgID}} if msg.OfflinePushInfo != nil { opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound diff --git a/internal/rpc/auth/auth.go b/internal/rpc/auth/auth.go index 62df74d214..a1acfd9313 100644 --- a/internal/rpc/auth/auth.go +++ b/internal/rpc/auth/auth.go @@ -16,6 +16,7 @@ package auth import ( "context" + "errors" "github.com/openimsdk/open-im-server/v3/pkg/common/config" redis2 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" @@ -66,6 +67,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg config.Share.Secret, config.RpcConfig.TokenPolicy.Expire, config.Share.MultiLogin, + config.Share.IMAdminUserID, ), config: config, }) @@ -129,6 +131,10 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim if err != nil { return nil, errs.Wrap(err) } + isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID) + if isAdmin { + return claims, nil + } m, err := s.authDatabase.GetTokensWithoutError(ctx, claims.UserID, claims.PlatformID) if err != nil { return nil, err @@ -190,7 +196,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID } m, err := s.authDatabase.GetTokensWithoutError(ctx, userID, int(platformID)) - if err != nil && err != redis.Nil { + if err != nil && errors.Is(err, redis.Nil) { return err } for k := range m { @@ -208,7 +214,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.InvalidateTokenReq) (*pbauth.InvalidateTokenResp, error) { m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID)) - if err != nil && err != redis.Nil { + if err != nil && errors.Is(err, redis.Nil) { return nil, err } if m == nil { diff --git a/internal/rpc/conversation/conversation.go b/internal/rpc/conversation/conversation.go index 6f6ca1f674..0c8a6fd857 100644 --- a/internal/rpc/conversation/conversation.go +++ b/internal/rpc/conversation/conversation.go @@ -261,27 +261,35 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver setConversationFieldsFunc := func() { if req.Conversation.RecvMsgOpt != nil { + conversation.RecvMsgOpt = req.Conversation.RecvMsgOpt.Value m["recv_msg_opt"] = req.Conversation.RecvMsgOpt.Value } if req.Conversation.AttachedInfo != nil { + conversation.AttachedInfo = req.Conversation.AttachedInfo.Value m["attached_info"] = req.Conversation.AttachedInfo.Value } if req.Conversation.Ex != nil { + conversation.Ex = req.Conversation.Ex.Value m["ex"] = req.Conversation.Ex.Value } if req.Conversation.IsPinned != nil { + conversation.IsPinned = req.Conversation.IsPinned.Value m["is_pinned"] = req.Conversation.IsPinned.Value } if req.Conversation.GroupAtType != nil { + conversation.GroupAtType = req.Conversation.GroupAtType.Value m["group_at_type"] = req.Conversation.GroupAtType.Value } if req.Conversation.MsgDestructTime != nil { + conversation.MsgDestructTime = req.Conversation.MsgDestructTime.Value m["msg_destruct_time"] = req.Conversation.MsgDestructTime.Value } if req.Conversation.IsMsgDestruct != nil { + conversation.IsMsgDestruct = req.Conversation.IsMsgDestruct.Value m["is_msg_destruct"] = req.Conversation.IsMsgDestruct.Value } if req.Conversation.BurnDuration != nil { + conversation.BurnDuration = req.Conversation.BurnDuration.Value m["burn_duration"] = req.Conversation.BurnDuration.Value } } diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index 30ac19a76a..b5ab1b2097 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -1811,7 +1811,6 @@ func (g *groupServer) GetSpecifiedUserGroupRequestInfo(ctx context.Context, req } if req.UserID != opUserID { - req.UserID = mcontext.GetOpUserID(ctx) adminIDs, err := g.db.GetGroupRoleLevelMemberIDs(ctx, req.GroupID, constant.GroupAdmin) if err != nil { return nil, err @@ -1820,10 +1819,11 @@ func (g *groupServer) GetSpecifiedUserGroupRequestInfo(ctx context.Context, req adminIDs = append(adminIDs, owners[0].UserID) adminIDs = append(adminIDs, g.config.Share.IMAdminUserID...) - if !datautil.Contain(req.UserID, adminIDs...) { + if !datautil.Contain(opUserID, adminIDs...) { return nil, errs.ErrNoPermission.WrapMsg("opUser no permission") } } + requests, err := g.db.FindGroupRequests(ctx, req.GroupID, []string{req.UserID}) if err != nil { return nil, err diff --git a/internal/rpc/msg/as_read.go b/internal/rpc/msg/as_read.go index 03f35b42d5..312c4d5562 100644 --- a/internal/rpc/msg/as_read.go +++ b/internal/rpc/msg/as_read.go @@ -16,6 +16,7 @@ package msg import ( "context" + "errors" cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" "github.com/openimsdk/protocol/constant" @@ -108,7 +109,7 @@ func (m *msgServer) MarkMsgsAsRead(ctx context.Context, req *msg.MarkMsgsAsReadR return nil, err } currentHasReadSeq, err := m.MsgDatabase.GetHasReadSeq(ctx, req.UserID, req.ConversationID) - if err != nil && errs.Unwrap(err) != redis.Nil { + if err != nil && errors.Is(err, redis.Nil) { return nil, err } if hasReadSeq > currentHasReadSeq { @@ -136,7 +137,7 @@ func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkCon return nil, err } hasReadSeq, err := m.MsgDatabase.GetHasReadSeq(ctx, req.UserID, req.ConversationID) - if err != nil && errs.Unwrap(err) != redis.Nil { + if err != nil && errors.Is(err, redis.Nil) { return nil, err } var seqs []int64 diff --git a/internal/rpc/msg/send.go b/internal/rpc/msg/send.go index 2c3f8c0a3b..f255d3f9ac 100644 --- a/internal/rpc/msg/send.go +++ b/internal/rpc/msg/send.go @@ -16,6 +16,7 @@ package msg import ( "context" + "github.com/openimsdk/tools/mw" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" @@ -78,8 +79,15 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq) } func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgData) { + log.ZDebug(nctx, "setConversationAtInfo", "msg", msg) + defer func() { + if r := recover(); r != nil { + mw.PanicStackToLog(nctx, r) + } + }() + ctx := mcontext.NewCtx("@@@" + mcontext.GetOperationID(nctx)) var atUserID []string diff --git a/internal/rpc/msg/seq.go b/internal/rpc/msg/seq.go index 5d40160de5..ddf84a267c 100644 --- a/internal/rpc/msg/seq.go +++ b/internal/rpc/msg/seq.go @@ -16,15 +16,15 @@ package msg import ( "context" + "errors" pbmsg "github.com/openimsdk/protocol/msg" - "github.com/openimsdk/tools/errs" "github.com/redis/go-redis/v9" "sort" ) func (m *msgServer) GetConversationMaxSeq(ctx context.Context, req *pbmsg.GetConversationMaxSeqReq) (*pbmsg.GetConversationMaxSeqResp, error) { maxSeq, err := m.MsgDatabase.GetMaxSeq(ctx, req.ConversationID) - if err != nil && errs.Unwrap(err) != redis.Nil { + if err != nil && errors.Is(err, redis.Nil) { return nil, err } return &pbmsg.GetConversationMaxSeqResp{MaxSeq: maxSeq}, nil diff --git a/internal/rpc/msg/server.go b/internal/rpc/msg/server.go index 91f41f1b11..536a73c833 100644 --- a/internal/rpc/msg/server.go +++ b/internal/rpc/msg/server.go @@ -139,3 +139,11 @@ func (m *msgServer) conversationAndGetRecvID(conversation *conversation.Conversa } return "" } + +func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) { + return nil, nil +} + +func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) { + return nil, nil +} diff --git a/internal/rpc/relation/friend.go b/internal/rpc/relation/friend.go index 9d55ba4d99..2f4843a8ef 100644 --- a/internal/rpc/relation/friend.go +++ b/internal/rpc/relation/friend.go @@ -273,7 +273,14 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *relation.SetFri return &relation.SetFriendRemarkResp{}, nil } -// ok. +func (s *friendServer) GetFriendInfo(ctx context.Context, req *relation.GetFriendInfoReq) (*relation.GetFriendInfoResp, error) { + friends, err := s.db.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs) + if err != nil { + return nil, err + } + return &relation.GetFriendInfoResp{FriendInfos: convert.FriendOnlyDB2PbOnly(friends)}, nil +} + func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *relation.GetDesignatedFriendsReq) (resp *relation.GetDesignatedFriendsResp, err error) { resp = &relation.GetDesignatedFriendsResp{} if datautil.Duplicate(req.FriendUserIDs) { diff --git a/internal/rpc/third/third.go b/internal/rpc/third/third.go index 0eeaaa3140..d37689b313 100644 --- a/internal/rpc/third/third.go +++ b/internal/rpc/third/third.go @@ -74,6 +74,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg if err != nil { return err } + // Select the oss method according to the profile policy enable := config.RpcConfig.Object.Enable var ( diff --git a/internal/rpc/third/tool.go b/internal/rpc/third/tool.go index ac4be39689..4e22ffbf97 100644 --- a/internal/rpc/third/tool.go +++ b/internal/rpc/third/tool.go @@ -82,3 +82,11 @@ func checkValidObjectName(objectName string) error { func (t *thirdServer) IsManagerUserID(opUserID string) bool { return authverify.IsManagerUserID(opUserID, t.config.Share.IMAdminUserID) } + +func putUpdate[T any](update map[string]any, name string, val interface{ GetValuePtr() *T }) { + ptrVal := val.GetValuePtr() + if ptrVal == nil { + return + } + update[name] = *ptrVal +} diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index da6c63d600..468a150e83 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -69,6 +69,7 @@ type Mongo struct { Database string `mapstructure:"database"` Username string `mapstructure:"username"` Password string `mapstructure:"password"` + AuthSource string `mapstructure:"authSource"` MaxPoolSize int `mapstructure:"maxPoolSize"` MaxRetry int `mapstructure:"maxRetry"` } @@ -212,12 +213,12 @@ type Push struct { FilePath string `mapstructure:"filePath"` AuthURL string `mapstructure:"authURL"` } `mapstructure:"fcm"` - JPNS struct { + JPush struct { AppKey string `mapstructure:"appKey"` MasterSecret string `mapstructure:"masterSecret"` PushURL string `mapstructure:"pushURL"` PushIntent string `mapstructure:"pushIntent"` - } `mapstructure:"jpns"` + } `mapstructure:"jpush"` IOSPush struct { PushSound string `mapstructure:"pushSound"` BadgeCount bool `mapstructure:"badgeCount"` @@ -368,20 +369,8 @@ type Share struct { } type MultiLogin struct { - Policy int `mapstructure:"policy"` - MaxNumOneEnd int `mapstructure:"maxNumOneEnd"` - CustomizeLoginNum struct { - IOS int `mapstructure:"ios"` - Android int `mapstructure:"android"` - Windows int `mapstructure:"windows"` - OSX int `mapstructure:"osx"` - Web int `mapstructure:"web"` - MiniWeb int `mapstructure:"miniWeb"` - Linux int `mapstructure:"linux"` - APad int `mapstructure:"aPad"` - IPad int `mapstructure:"iPad"` - Admin int `mapstructure:"admin"` - } `mapstructure:"customizeLoginNum"` + Policy int `mapstructure:"policy"` + MaxNumOneEnd int `mapstructure:"maxNumOneEnd"` } type RpcRegisterName struct { @@ -490,6 +479,7 @@ func (m *Mongo) Build() *mongoutil.Config { Database: m.Database, Username: m.Username, Password: m.Password, + AuthSource: m.AuthSource, MaxPoolSize: m.MaxPoolSize, MaxRetry: m.MaxRetry, } diff --git a/pkg/common/convert/friend.go b/pkg/common/convert/friend.go index 8d6cfad183..6d346b0f4a 100644 --- a/pkg/common/convert/friend.go +++ b/pkg/common/convert/friend.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" + "github.com/openimsdk/protocol/relation" "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/utils/datautil" @@ -35,9 +36,7 @@ func FriendPb2DB(friend *sdkws.FriendInfo) *model.Friend { return dbFriend } -func FriendDB2Pb(ctx context.Context, friendDB *model.Friend, - getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error), -) (*sdkws.FriendInfo, error) { +func FriendDB2Pb(ctx context.Context, friendDB *model.Friend, getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error)) (*sdkws.FriendInfo, error) { users, err := getUsers(ctx, []string{friendDB.FriendUserID}) if err != nil { return nil, err @@ -53,11 +52,7 @@ func FriendDB2Pb(ctx context.Context, friendDB *model.Friend, }, nil } -func FriendsDB2Pb( - ctx context.Context, - friendsDB []*model.Friend, - getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error), -) (friendsPb []*sdkws.FriendInfo, err error) { +func FriendsDB2Pb(ctx context.Context, friendsDB []*model.Friend, getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error)) (friendsPb []*sdkws.FriendInfo, err error) { if len(friendsDB) == 0 { return nil, nil } @@ -86,7 +81,21 @@ func FriendsDB2Pb( friendsPb = append(friendsPb, friendPb) } return friendsPb, nil +} +func FriendOnlyDB2PbOnly(friendsDB []*model.Friend) []*relation.FriendInfoOnly { + return datautil.Slice(friendsDB, func(f *model.Friend) *relation.FriendInfoOnly { + return &relation.FriendInfoOnly{ + OwnerUserID: f.OwnerUserID, + FriendUserID: f.FriendUserID, + Remark: f.Remark, + CreateTime: f.CreateTime.UnixMilli(), + AddSource: f.AddSource, + OperatorUserID: f.OperatorUserID, + Ex: f.Ex, + IsPinned: f.IsPinned, + } + }) } func FriendRequestDB2Pb(ctx context.Context, friendRequests []*model.FriendRequest, getUsers func(ctx context.Context, userIDs []string) (map[string]*sdkws.UserInfo, error)) ([]*sdkws.FriendRequest, error) { diff --git a/pkg/common/storage/cache/redis/msg.go b/pkg/common/storage/cache/redis/msg.go index 30f367bb75..b04bc5c357 100644 --- a/pkg/common/storage/cache/redis/msg.go +++ b/pkg/common/storage/cache/redis/msg.go @@ -1,17 +1,3 @@ -// Copyright © 2023 OpenIM. All rights reserved. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - package redis import ( diff --git a/pkg/common/storage/controller/auth.go b/pkg/common/storage/controller/auth.go index de8f93462f..b134c3c3ba 100644 --- a/pkg/common/storage/controller/auth.go +++ b/pkg/common/storage/controller/auth.go @@ -2,15 +2,15 @@ package controller import ( "context" - "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey" - "github.com/openimsdk/tools/log" "github.com/golang-jwt/jwt/v4" "github.com/openimsdk/open-im-server/v3/pkg/authverify" + "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/protocol/constant" "github.com/openimsdk/tools/errs" + "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/tokenverify" ) @@ -26,9 +26,8 @@ type AuthDatabase interface { } type multiLoginConfig struct { - Policy int - MaxNumOneEnd int - CustomizeLoginNum map[int]int + Policy int + MaxNumOneEnd int } type authDatabase struct { @@ -36,25 +35,15 @@ type authDatabase struct { accessSecret string accessExpire int64 multiLogin multiLoginConfig + adminUserIDs []string } -func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin) AuthDatabase { +func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin, adminUserIDs []string) AuthDatabase { return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLogin: multiLoginConfig{ Policy: multiLogin.Policy, MaxNumOneEnd: multiLogin.MaxNumOneEnd, - CustomizeLoginNum: map[int]int{ - constant.IOSPlatformID: multiLogin.CustomizeLoginNum.IOS, - constant.AndroidPlatformID: multiLogin.CustomizeLoginNum.Android, - constant.WindowsPlatformID: multiLogin.CustomizeLoginNum.Windows, - constant.OSXPlatformID: multiLogin.CustomizeLoginNum.OSX, - constant.WebPlatformID: multiLogin.CustomizeLoginNum.Web, - constant.MiniWebPlatformID: multiLogin.CustomizeLoginNum.MiniWeb, - constant.LinuxPlatformID: multiLogin.CustomizeLoginNum.Linux, - constant.AndroidPadPlatformID: multiLogin.CustomizeLoginNum.APad, - constant.IPadPlatformID: multiLogin.CustomizeLoginNum.IPad, - constant.AdminPlatformID: multiLogin.CustomizeLoginNum.Admin, - }, - }} + }, adminUserIDs: adminUserIDs, + } } // If the result is empty. @@ -91,27 +80,31 @@ func (a *authDatabase) BatchSetTokenMapByUidPid(ctx context.Context, tokens []st // Create Token. func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) { - tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID) - if err != nil { - return "", err - } - deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID) - if err != nil { - return "", err - } - if len(deleteTokenKey) != 0 { - err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey) + isAdmin := authverify.IsManagerUserID(userID, a.adminUserIDs) + if !isAdmin { + tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID) if err != nil { return "", err } - } - if len(kickedTokenKey) != 0 { - for _, k := range kickedTokenKey { - err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken) + + deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID) + if err != nil { + return "", err + } + if len(deleteTokenKey) != 0 { + err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey) if err != nil { return "", err } - log.ZDebug(ctx, "kicked token in create token", "token", k) + } + if len(kickedTokenKey) != 0 { + for _, k := range kickedTokenKey { + err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken) + if err != nil { + return "", err + } + log.ZDebug(ctx, "kicked token in create token", "token", k) + } } } @@ -122,9 +115,12 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI return "", errs.WrapMsg(err, "token.SignedString") } - if err = a.cache.SetTokenFlagEx(ctx, userID, platformID, tokenString, constant.NormalToken); err != nil { - return "", err + if !isAdmin { + if err = a.cache.SetTokenFlagEx(ctx, userID, platformID, tokenString, constant.NormalToken); err != nil { + return "", err + } } + return tokenString, nil } @@ -172,17 +168,8 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string kickToken = append(kickToken, ts[len(ts)-1]) } } - case constant.SingleTerminalLogin: - for _, ts := range loginTokenMap { - kickToken = append(kickToken, ts...) - } - case constant.WebAndOther: - unkickTerminal = constant.WebPlatformStr - fallthrough case constant.PCAndOther: - if unkickTerminal == "" { - unkickTerminal = constant.TerminalPC - } + unkickTerminal = constant.TerminalPC if constant.PlatformIDToClass(platformID) != unkickTerminal { for plt, ts := range loginTokenMap { if constant.PlatformIDToClass(plt) != unkickTerminal { @@ -214,17 +201,17 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string } } } - case constant.PcMobileAndWeb: + case constant.AllLoginButSameClassKick: var ( - reserved = make(map[string]bool) + reserved = make(map[string]struct{}) ) for plt, ts := range loginTokenMap { if constant.PlatformIDToClass(plt) == constant.PlatformIDToClass(platformID) { kickToken = append(kickToken, ts...) } else { - if !reserved[constant.PlatformIDToClass(plt)] { - reserved[constant.PlatformIDToClass(plt)] = true + if _, ok := reserved[constant.PlatformIDToClass(plt)]; !ok { + reserved[constant.PlatformIDToClass(plt)] = struct{}{} kickToken = append(kickToken, ts[:len(ts)-1]...) continue } else { @@ -232,36 +219,20 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string } } } - - case constant.Customize: - if a.multiLogin.CustomizeLoginNum[platformID] <= 0 { - return nil, nil, errs.New("Do not allow login on this end").Wrap() - } - for plt, ts := range loginTokenMap { - l := len(ts) - if platformID == plt { - l++ - } - // a.multiLogin.CustomizeLoginNum[platformID] must > 0 - limit := min(a.multiLogin.CustomizeLoginNum[plt], a.multiLogin.MaxNumOneEnd) - if l > limit { - kickToken = append(kickToken, ts[:l-limit]...) - } - } default: return nil, nil, errs.New("unknown multiLogin policy").Wrap() } - var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd - if a.multiLogin.Policy == constant.Customize { - adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID] - } - l := len(adminToken) - if platformID == constant.AdminPlatformID { - l++ - } - if l > adminTokenMaxNum { - kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...) - } + //var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd + //if a.multiLogin.Policy == constant.Customize { + // adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID] + //} + //l := len(adminToken) + //if platformID == constant.AdminPlatformID { + // l++ + //} + //if l > adminTokenMaxNum { + // kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...) + //} return deleteToken, kickToken, nil } diff --git a/pkg/common/storage/controller/msg.go b/pkg/common/storage/controller/msg.go index d579069b69..789adb1f66 100644 --- a/pkg/common/storage/controller/msg.go +++ b/pkg/common/storage/controller/msg.go @@ -372,7 +372,7 @@ func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID strin // This ensures that their message retrieval starts from the point they joined. func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (int64, int64, []*sdkws.MsgData, error) { userMinSeq, err := db.seqUser.GetUserMinSeq(ctx, conversationID, userID) - if err != nil && errs.Unwrap(err) != redis.Nil { + if err != nil && errors.Is(err, redis.Nil) { return 0, 0, nil, err } minSeq, err := db.seqConversation.GetMinSeq(ctx, conversationID) @@ -443,6 +443,11 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin return 0, 0, nil, err } successMsgs = append(mongoMsgs, successMsgs...) + + _, err = db.msg.SetMessagesToCache(ctx, conversationID, mongoMsgs) + if err != nil { + return 0, 0, nil, err + } } return minSeq, maxSeq, successMsgs, nil @@ -485,7 +490,7 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co } successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs) if err != nil { - if err != redis.Nil { + if errors.Is(err, redis.Nil) { log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID) } } @@ -500,6 +505,11 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co } successMsgs = append(successMsgs, mongoMsgs...) + + _, err = db.msg.SetMessagesToCache(ctx, conversationID, mongoMsgs) + if err != nil { + return 0, 0, nil, err + } } return minSeq, maxSeq, successMsgs, nil } diff --git a/pkg/common/storage/database/mgo/group.go b/pkg/common/storage/database/mgo/group.go index 3be7883af9..620269b43f 100644 --- a/pkg/common/storage/database/mgo/group.go +++ b/pkg/common/storage/database/mgo/group.go @@ -76,7 +76,7 @@ func (g *GroupMgo) Take(ctx context.Context, groupID string) (group *model.Group func (g *GroupMgo) Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, groups []*model.Group, err error) { // Define the sorting options - opts := options.Find().SetSort(bson.D{{Key: "created_at", Value: -1}}) + opts := options.Find().SetSort(bson.D{{Key: "create_time", Value: -1}}) // Perform the search with pagination and sorting return mongoutil.FindPage[*model.Group](ctx, g.coll, bson.M{ diff --git a/pkg/common/storage/model/application.go b/pkg/common/storage/model/application.go new file mode 100644 index 0000000000..b09b0e8948 --- /dev/null +++ b/pkg/common/storage/model/application.go @@ -0,0 +1,18 @@ +package model + +import ( + "go.mongodb.org/mongo-driver/bson/primitive" + "time" +) + +type Application struct { + ID primitive.ObjectID `bson:"_id"` + Platform string `bson:"platform"` + Hot bool `bson:"hot"` + Version string `bson:"version"` + Url string `bson:"url"` + Text string `bson:"text"` + Force bool `bson:"force"` + Latest bool `bson:"latest"` + CreateTime time.Time `bson:"create_time"` +} diff --git a/pkg/rpccache/subscriber.go b/pkg/rpccache/subscriber.go index 3046f84b11..3c73ef4496 100644 --- a/pkg/rpccache/subscriber.go +++ b/pkg/rpccache/subscriber.go @@ -17,12 +17,18 @@ package rpccache import ( "context" "encoding/json" + "github.com/openimsdk/tools/mw" "github.com/openimsdk/tools/log" "github.com/redis/go-redis/v9" ) func subscriberRedisDeleteCache(ctx context.Context, client redis.UniversalClient, channel string, del func(ctx context.Context, key ...string)) { + defer func() { + if r := recover(); r != nil { + mw.PanicStackToLog(ctx, r) + } + }() for message := range client.Subscribe(ctx, channel).Channel() { log.ZDebug(ctx, "subscriberRedisDeleteCache", "channel", channel, "payload", message.Payload) var keys []string diff --git a/scripts/create-topic.sh b/scripts/create-topic.sh deleted file mode 100755 index bbc739287f..0000000000 --- a/scripts/create-topic.sh +++ /dev/null @@ -1,55 +0,0 @@ -#!/usr/bin/env bash -# Copyright © 2023 OpenIM. All rights reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# Wait for Kafka to be ready - -KAFKA_SERVER=localhost:9092 - -MAX_ATTEMPTS=300 -attempt_num=1 - -echo "Waiting for Kafka to be ready..." - -until /opt/bitnami/kafka/bin/kafka-topics.sh --list --bootstrap-server $KAFKA_SERVER; do - echo "Attempt $attempt_num of $MAX_ATTEMPTS: Kafka not ready yet..." - if [ $attempt_num -eq $MAX_ATTEMPTS ]; then - echo "Kafka not ready after $MAX_ATTEMPTS attempts, exiting" - exit 1 - fi - attempt_num=$((attempt_num+1)) - sleep 1 -done - -echo "Kafka is ready. Creating topics..." - - -topics=("toRedis" "toMongo" "toPush" "toOfflinePush") -partitions=8 -replicationFactor=1 - -for topic in "${topics[@]}"; do - if /opt/bitnami/kafka/bin/kafka-topics.sh --create \ - --bootstrap-server $KAFKA_SERVER \ - --replication-factor $replicationFactor \ - --partitions $partitions \ - --topic $topic - then - echo "Topic $topic created." - else - echo "Failed to create topic $topic." - fi -done - -echo "All topics created." diff --git a/scripts/mongo-init.sh b/scripts/mongo-init.sh deleted file mode 100755 index 25bb2d654c..0000000000 --- a/scripts/mongo-init.sh +++ /dev/null @@ -1,66 +0,0 @@ -# Copyright © 2023 OpenIM. All rights reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -mongosh < 0) { - try { - db = connect('mongodb://127.0.0.1:27017/admin'); - var authResult = db.auth(rootUsername, rootPassword); - if (authResult) { - print('Authentication successful for root user: ' + rootUsername); - connected = true; - } else { - print('Authentication failed for root user: ' + rootUsername + ' with password: ' + rootPassword); - quit(1); - } - } catch (e) { - maxRetries--; - print('Connection failed, retrying... Remaining attempts: ' + maxRetries); - sleep(1000); // Sleep for 1 second - } -} - -if (connected) { - db = db.getSiblingDB(dbName); - var createUserResult = db.createUser({ - user: openimUsername, - pwd: openimPassword, - roles: [{ - role: 'readWrite', - db: dbName - }] - }); - - if (createUserResult.ok == 1) { - print('User creation successful. User: ' + openimUsername + ', Database: ' + dbName); - } else { - print('User creation failed for user: ' + openimUsername + ' in database: ' + dbName); - quit(1); - } -} else { - print('Failed to connect to MongoDB after 300 retries.'); - quit(1); -} -EOF - - - - diff --git a/tools/changelog/changelog.go b/tools/changelog/changelog.go new file mode 100644 index 0000000000..75d914a279 --- /dev/null +++ b/tools/changelog/changelog.go @@ -0,0 +1,198 @@ +package main + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "regexp" + "strings" +) + +// You can specify a tag as a command line argument to generate the changelog for a specific version. +// Example: go run tools/changelog/changelog.go v0.0.33 +// If no tag is provided, the latest release will be used. + +// Setting repo owner and repo name by generate changelog +const ( + repoOwner = "openimsdk" + repoName = "open-im-server" +) + +// GitHubRepo struct represents the repo details. +type GitHubRepo struct { + Owner string + Repo string + FullChangelog string +} + +// ReleaseData represents the JSON structure for release data. +type ReleaseData struct { + TagName string `json:"tag_name"` + Body string `json:"body"` + HtmlUrl string `json:"html_url"` + Published string `json:"published_at"` +} + +// Method to classify and format release notes. +func (g *GitHubRepo) classifyReleaseNotes(body string) map[string][]string { + result := map[string][]string{ + "feat": {}, + "fix": {}, + "chore": {}, + "refactor": {}, + "build": {}, + "other": {}, + } + + // Regular expression to extract PR number and URL (case insensitive) + rePR := regexp.MustCompile(`(?i)in (https://github\.com/[^\s]+/pull/(\d+))`) + + // Split the body into individual lines. + lines := strings.Split(body, "\n") + + for _, line := range lines { + // Skip lines that contain "deps: Merge" + if strings.Contains(strings.ToLower(line), "deps: merge #") { + continue + } + + // Use a regular expression to extract Full Changelog link and its title (case insensitive). + if strings.Contains(strings.ToLower(line), "**full changelog**") { + matches := regexp.MustCompile(`(?i)\*\*full changelog\*\*: (https://github\.com/[^\s]+/compare/([^\s]+))`).FindStringSubmatch(line) + if len(matches) > 2 { + // Format the Full Changelog link with title + g.FullChangelog = fmt.Sprintf("[%s](%s)", matches[2], matches[1]) + } + continue // Skip further processing for this line. + } + + if strings.HasPrefix(line, "*") { + var category string + + // Use strings.ToLower to make the matching case insensitive + lowerLine := strings.ToLower(line) + + // Determine the category based on the prefix (case insensitive). + if strings.HasPrefix(lowerLine, "* feat") { + category = "feat" + } else if strings.HasPrefix(lowerLine, "* fix") { + category = "fix" + } else if strings.HasPrefix(lowerLine, "* chore") { + category = "chore" + } else if strings.HasPrefix(lowerLine, "* refactor") { + category = "refactor" + } else if strings.HasPrefix(lowerLine, "* build") { + category = "build" + } else { + category = "other" + } + + // Extract PR number and URL (case insensitive) + matches := rePR.FindStringSubmatch(line) + if len(matches) == 3 { + prURL := matches[1] + prNumber := matches[2] + // Format the line with the PR link and use original content for the final result + formattedLine := fmt.Sprintf("* %s [#%s](%s)", strings.Split(line, " by ")[0][2:], prNumber, prURL) + result[category] = append(result[category], formattedLine) + } else { + // If no PR link is found, just add the line as is + result[category] = append(result[category], line) + } + } + } + + return result +} + +// Method to generate the final changelog. +func (g *GitHubRepo) generateChangelog(tag, date, htmlURL, body string) string { + sections := g.classifyReleaseNotes(body) + + // Convert ISO 8601 date to simpler format (YYYY-MM-DD) + formattedDate := date[:10] + + // Changelog header with tag, date, and links. + changelog := fmt.Sprintf("## [%s](%s) \t(%s)\n\n", tag, htmlURL, formattedDate) + + if len(sections["feat"]) > 0 { + changelog += "### New Features\n" + strings.Join(sections["feat"], "\n") + "\n\n" + } + if len(sections["fix"]) > 0 { + changelog += "### Bug Fixes\n" + strings.Join(sections["fix"], "\n") + "\n\n" + } + if len(sections["chore"]) > 0 { + changelog += "### Chores\n" + strings.Join(sections["chore"], "\n") + "\n\n" + } + if len(sections["refactor"]) > 0 { + changelog += "### Refactors\n" + strings.Join(sections["refactor"], "\n") + "\n\n" + } + if len(sections["build"]) > 0 { + changelog += "### Builds\n" + strings.Join(sections["build"], "\n") + "\n\n" + } + if len(sections["other"]) > 0 { + changelog += "### Others\n" + strings.Join(sections["other"], "\n") + "\n\n" + } + + if g.FullChangelog != "" { + changelog += fmt.Sprintf("**Full Changelog**: %s\n", g.FullChangelog) + } + + return changelog +} + +// Method to fetch release data from GitHub API. +func (g *GitHubRepo) fetchReleaseData(version string) (*ReleaseData, error) { + var apiURL string + + if version == "" { + // Fetch the latest release. + apiURL = fmt.Sprintf("https://api.github.com/repos/%s/%s/releases/latest", g.Owner, g.Repo) + } else { + // Fetch a specific version. + apiURL = fmt.Sprintf("https://api.github.com/repos/%s/%s/releases/tags/%s", g.Owner, g.Repo, version) + } + + resp, err := http.Get(apiURL) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + var releaseData ReleaseData + err = json.Unmarshal(body, &releaseData) + if err != nil { + return nil, err + } + + return &releaseData, nil +} + +func main() { + repo := &GitHubRepo{Owner: repoOwner, Repo: repoName} + + // Get the version from command line arguments, if provided + var version string // Default is use latest + + if len(os.Args) > 1 { + version = os.Args[1] // Use the provided version + } + + // Fetch release data (either for latest or specific version) + releaseData, err := repo.fetchReleaseData(version) + if err != nil { + fmt.Println("Error fetching release data:", err) + return + } + + // Generate and print the formatted changelog + changelog := repo.generateChangelog(releaseData.TagName, releaseData.Published, releaseData.HtmlUrl, releaseData.Body) + fmt.Println(changelog) +} diff --git a/tools/changelog/main.go b/tools/changelog/main.go deleted file mode 100644 index ff9a7eab9b..0000000000 --- a/tools/changelog/main.go +++ /dev/null @@ -1,308 +0,0 @@ -// Copyright © 2023 OpenIM. All rights reserved. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import ( - "fmt" - "log" - "os" - "os/exec" - "regexp" - "sort" - "strings" -) - -var ( - mergeRequest = regexp.MustCompile(`Merge pull request #([\d]+)`) - webconsoleBump = regexp.MustCompile(regexp.QuoteMeta("bump(github.com/openshift/origin-web-console): ") + `([\w]+)`) - upstreamKube = regexp.MustCompile(`^UPSTREAM: (\d+)+:(.+)`) - upstreamRepo = regexp.MustCompile(`^UPSTREAM: ([\w/-]+): (\d+)+:(.+)`) - prefix = regexp.MustCompile(`^[\w-]: `) - - assignments = []prefixAssignment{ - {"cluster up", "cluster"}, - {" pv ", "storage"}, - {"haproxy", "router"}, - {"router", "router"}, - {"route", "route"}, - {"authoriz", "auth"}, - {"rbac", "auth"}, - {"authent", "auth"}, - {"reconcil", "auth"}, - {"auth", "auth"}, - {"role", "auth"}, - {" dc ", "deploy"}, - {"deployment", "deploy"}, - {"rolling", "deploy"}, - {"security context constr", "security"}, - {"scc", "security"}, - {"pipeline", "build"}, - {"build", "build"}, - {"registry", "registry"}, - {"registries", "image"}, - {"image", "image"}, - {" arp ", "network"}, - {" cni ", "network"}, - {"egress", "network"}, - {"network", "network"}, - {"oc ", "cli"}, - {"template", "template"}, - {"etcd", "server"}, - {"pod", "node"}, - {"scripts/", "hack"}, - {"e2e", "test"}, - {"integration", "test"}, - {"cluster", "cluster"}, - {"master", "server"}, - {"packages", "hack"}, - {"api", "server"}, - } -) - -type prefixAssignment struct { - term string - prefix string -} - -type commit struct { - short string - parents []string - message string -} - -func contains(arr []string, value string) bool { - for _, s := range arr { - if s == value { - return true - } - } - return false -} - -func main() { - log.SetFlags(0) - if len(os.Args) != 3 { - log.Fatalf("Must specify two arguments, FROM and TO") - } - from := os.Args[1] - to := os.Args[2] - - out, err := exec.Command("git", "log", "--topo-order", "--pretty=tformat:%h %p|%s", "--reverse", fmt.Sprintf("%s..%s", from, to)).CombinedOutput() - if err != nil { - log.Fatal(err) - } - - hide := make(map[string]struct{}) - var apiChanges []string - var webconsole []string - var commits []commit - var upstreams []commit - var bumps []commit - for _, line := range strings.Split(string(out), "\n") { - if len(strings.TrimSpace(line)) == 0 { - continue - } - parts := strings.SplitN(line, "|", 2) - hashes := strings.Split(parts[0], " ") - c := commit{short: hashes[0], parents: hashes[1:], message: parts[1]} - - if strings.HasPrefix(c.message, "UPSTREAM: ") { - hide[c.short] = struct{}{} - upstreams = append(upstreams, c) - } - if strings.HasPrefix(c.message, "bump(") { - hide[c.short] = struct{}{} - bumps = append(bumps, c) - } - - if len(c.parents) == 1 { - commits = append(commits, c) - continue - } - - matches := mergeRequest.FindStringSubmatch(line) - if len(matches) == 0 { - // this may have been a human pressing the merge button, we'll just record this as a direct push - continue - } - - // split the accumulated commits into any that are force merges (assumed to be the initial set due - // to --topo-order) from the PR commits as soon as we see any of our merge parents. Then print - // any of the force merges - var first int - for i := range commits { - first = i - if contains(c.parents, commits[i].short) { - first++ - break - } - } - individual := commits[:first] - merged := commits[first:] - for _, commit := range individual { - if len(commit.parents) > 1 { - continue - } - if _, ok := hide[commit.short]; ok { - continue - } - fmt.Printf("force-merge: %s %s\n", commit.message, commit.short) - } - - // try to find either the PR title or the first commit title from the merge commit - out, err := exec.Command("git", "show", "--pretty=tformat:%b", c.short).CombinedOutput() - if err != nil { - log.Fatal(err) - } - var message string - para := strings.Split(string(out), "\n\n") - if len(para) > 0 && strings.HasPrefix(para[0], "Automatic merge from submit-queue") { - para = para[1:] - } - // this is no longer necessary with the submit queue in place - if len(para) > 0 && strings.HasPrefix(para[0], "Merged by ") { - para = para[1:] - } - // post submit-queue, the merge bot will add the PR title, which is usually pretty good - if len(para) > 0 { - message = strings.Split(para[0], "\n")[0] - } - if len(message) == 0 && len(merged) > 0 { - message = merged[0].message - } - if len(message) > 0 && len(merged) == 1 && message == merged[0].message { - merged = nil - } - - // try to calculate a prefix based on the diff - if len(message) > 0 && !prefix.MatchString(message) { - prefix, ok := findPrefixFor(message, merged) - if ok { - message = prefix + ": " + message - } - } - - // github merge - - // has api changes - display := fmt.Sprintf("%s [\\#%s](https://github.com/openimsdk/Open-IM-Server/pull/%s)", message, matches[1], matches[1]) - if hasFileChanges(c.short, "pkg/apistruct/") { - apiChanges = append(apiChanges, display) - } - - var filtered []commit - for _, commit := range merged { - if _, ok := hide[commit.short]; ok { - continue - } - filtered = append(filtered, commit) - } - if len(filtered) > 0 { - fmt.Printf("- %s\n", display) - for _, commit := range filtered { - fmt.Printf(" - %s (%s)\n", commit.message, commit.short) - } - } - - // stick the merge commit in at the beginning of the next list so we can anchor the previous parent - commits = []commit{c} - } - - // chunk the bumps - var lines []string - for _, commit := range bumps { - if m := webconsoleBump.FindStringSubmatch(commit.message); len(m) > 0 { - webconsole = append(webconsole, m[1]) - continue - } - lines = append(lines, commit.message) - } - lines = sortAndUniq(lines) - for _, line := range lines { - fmt.Printf("- %s\n", line) - } - - // chunk the upstreams - lines = nil - for _, commit := range upstreams { - lines = append(lines, commit.message) - } - lines = sortAndUniq(lines) - for _, line := range lines { - fmt.Printf("- %s\n", upstreamLinkify(line)) - } - - if len(webconsole) > 0 { - fmt.Printf("- web: from %s^..%s\n", webconsole[0], webconsole[len(webconsole)-1]) - } - - for _, apiChange := range apiChanges { - fmt.Printf(" - %s\n", apiChange) - } -} - -func findPrefixFor(message string, commits []commit) (string, bool) { - message = strings.ToLower(message) - for _, m := range assignments { - if strings.Contains(message, m.term) { - return m.prefix, true - } - } - for _, c := range commits { - if prefix, ok := findPrefixFor(c.message, nil); ok { - return prefix, ok - } - } - return "", false -} - -func hasFileChanges(commit string, prefixes ...string) bool { - out, err := exec.Command("git", "diff", "--name-only", fmt.Sprintf("%s^..%s", commit, commit)).CombinedOutput() - if err != nil { - log.Fatal(err) - } - for _, file := range strings.Split(string(out), "\n") { - for _, prefix := range prefixes { - if strings.HasPrefix(file, prefix) { - return true - } - } - } - return false -} - -func sortAndUniq(lines []string) []string { - sort.Strings(lines) - out := make([]string, 0, len(lines)) - last := "" - for _, s := range lines { - if last == s { - continue - } - last = s - out = append(out, s) - } - return out -} - -func upstreamLinkify(line string) string { - if m := upstreamKube.FindStringSubmatch(line); len(m) > 0 { - return fmt.Sprintf("UPSTREAM: [#%s](https://github.com/openimsdk/open-im-server/pull/%s):%s", m[1], m[1], m[2]) - } - if m := upstreamRepo.FindStringSubmatch(line); len(m) > 0 { - return fmt.Sprintf("UPSTREAM: [%s#%s](https://github.com/%s/pull/%s):%s", m[1], m[2], m[1], m[2], m[3]) - } - return line -} diff --git a/tools/check-component/main.go b/tools/check-component/main.go index 4f4c08c16a..94dbd613c4 100644 --- a/tools/check-component/main.go +++ b/tools/check-component/main.go @@ -66,7 +66,7 @@ func CheckMinIO(ctx context.Context, config *config.Minio) error { } func CheckKafka(ctx context.Context, conf *config.Kafka) error { - return kafka.Check(ctx, conf.Build(), []string{conf.ToMongoTopic, conf.ToRedisTopic, conf.ToPushTopic, conf.ToOfflinePushTopic}) + return kafka.CheckHealth(ctx, conf.Build()) } func initConfig(configDir string) (*config.Mongo, *config.Redis, *config.Kafka, *config.Minio, *config.Discovery, error) {