deps: Merge #2804 #2805 #2806 #2808 PRs into pre-release-v3.8.2 (#2812)

* fix: improve time condition check mehtod. (#2804)

* fix: improve time condition check mehtod.

* fix

* fix: webhook before online push (#2805)

* fix: solve err Notification when setGroupInfo. (#2806)

* fix: solve err Notification when setGroupInfo.

* build: update checkout version.

* fix: update notification contents.

* fix: set own read seq in MongoDB when sender send a message. (#2808)

---------

Co-authored-by: Monet Lee <monet_lee@163.com>
Co-authored-by: icey-yu <119291641+icey-yu@users.noreply.github.com>
Co-authored-by: OpenIM-Gordon <46924906+FGadvancer@users.noreply.github.com>

.env

@@ -1,5 +1,6 @@
-MONGO_IMAGE=mongo:7.0
+MONGO_IMAGE=mongo:6.0.2
 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
@@ -15,3 +16,4 @@ 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=./
+

.github/workflows/changelog.yml

@@ -1,78 +0,0 @@
-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

.github/workflows/go-build-test.yml

@@ -149,7 +149,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go_version: ["1.22"]
+        go_version: ["1.21"]
 
     steps:
       - name: Checkout Repository

.github/workflows/update-version-file-on-release.yml

@@ -1,84 +0,0 @@
-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.");
-            }

CHANGELOG/CHANGELOG-3.8.md

@@ -1,75 +0,0 @@
-## [v3.8.2](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.2) 	(2024-11-25)
-
-### New Features
-* feat: improve publish docker image workflows [#2697](https://github.com/openimsdk/open-im-server/pull/2697)
-* feat: Msg filter [#2703](https://github.com/openimsdk/open-im-server/pull/2703)
-* feat: provide the interface required [#2712](https://github.com/openimsdk/open-im-server/pull/2712)
-* feat: add webhooks of online status and remove zookeeper configuration. [#2716](https://github.com/openimsdk/open-im-server/pull/2716)
-* feat: Add More Multi Login Policy [#2770](https://github.com/openimsdk/open-im-server/pull/2770)
-* feat: Push configuration can ignore case sensitivity [#2775](https://github.com/openimsdk/open-im-server/pull/2775)
-* feat: support app update service [#2794](https://github.com/openimsdk/open-im-server/pull/2794)
-* feat: implement merge milestone PR to target-branch. [#2796](https://github.com/openimsdk/open-im-server/pull/2796)
-* feat: support app update service [#2811](https://github.com/openimsdk/open-im-server/pull/2811)
-* feat: ApplicationVersion move chat [#2813](https://github.com/openimsdk/open-im-server/pull/2813)
-* feat: Update login policy [#2822](https://github.com/openimsdk/open-im-server/pull/2822)
-* feat: merge js sdk [#2856](https://github.com/openimsdk/open-im-server/pull/2856)
-* feat: Print Panic Log [#2850](https://github.com/openimsdk/open-im-server/pull/2850)
-
-### Bug Fixes
-* fix: update load file logic. [#2700](https://github.com/openimsdk/open-im-server/pull/2700)
-* fix: the message I sent is not set to read seq in mongodb [#2718](https://github.com/openimsdk/open-im-server/pull/2718)
-* fix: cannot modify group member avatars [#2719](https://github.com/openimsdk/open-im-server/pull/2719)
-* fix: auth package import twice [#2724](https://github.com/openimsdk/open-im-server/pull/2724)
-* fix: join the group chat directly, notification type error [#2772](https://github.com/openimsdk/open-im-server/pull/2772)
-* fix: change update group member level logic [#2730](https://github.com/openimsdk/open-im-server/pull/2730)
-* fix: joinSource check args error. [#2773](https://github.com/openimsdk/open-im-server/pull/2773)
-* fix: Change group member roleLevel can`t send notification [#2777](https://github.com/openimsdk/open-im-server/pull/2777)
-* fix: client sends message status error to server [#2779](https://github.com/openimsdk/open-im-server/pull/2779)
-* fix: del UserB's conversation version cache when userA set conversati… [#2785](https://github.com/openimsdk/open-im-server/pull/2785)
-* fix: improve setConversationAtInfo logic. [#2782](https://github.com/openimsdk/open-im-server/pull/2782)
-* fix: improve transfer Owner logic when newOwner is mute. [#2790](https://github.com/openimsdk/open-im-server/pull/2790)
-* fix: improve getUserInfo logic. [#2792](https://github.com/openimsdk/open-im-server/pull/2792)
-* fix: improve time condition check mehtod. [#2804](https://github.com/openimsdk/open-im-server/pull/2804)
-* fix: webhook before online push [#2805](https://github.com/openimsdk/open-im-server/pull/2805)
-* fix: set own read seq in MongoDB when sender send a message. [#2808](https://github.com/openimsdk/open-im-server/pull/2808)
-* fix: solve err Notification when setGroupInfo. [#2806](https://github.com/openimsdk/open-im-server/pull/2806)
-* fix: improve condition check. [#2815](https://github.com/openimsdk/open-im-server/pull/2815)
-* fix: Write back message to Redis [#2836](https://github.com/openimsdk/open-im-server/pull/2836)
-* fix: get group return repeated result [#2842](https://github.com/openimsdk/open-im-server/pull/2842)
-* fix: SetConversations can update new conversation [#2838](https://github.com/openimsdk/open-im-server/pull/2838)
-* fix(push): push content with jpush [#2844](https://github.com/openimsdk/open-im-server/pull/2844)
-* fix #2860 migrate jpns to jpush [#2861](https://github.com/openimsdk/open-im-server/pull/2861)
-* fix: concurrent write to websocket connection [#2866](https://github.com/openimsdk/open-im-server/pull/2866)
-* fix: Remove admin token in redis [#2871](https://github.com/openimsdk/open-im-server/pull/2871)
-* Fix Push2User webhookBeforeOfflinePush [#2862](https://github.com/openimsdk/open-im-server/pull/2862)
-* fix: move workflow to correct path [#2837](https://github.com/openimsdk/open-im-server/pull/2837)
-* fix: del login Policy [#2825](https://github.com/openimsdk/open-im-server/pull/2825)
-* fix: Wrong Redis Error Check [#2876](https://github.com/openimsdk/open-im-server/pull/2876)
-* fix: minor log typo [#2881](https://github.com/openimsdk/open-im-server/pull/2881)
-* fix: webhookAfterSingleMsgRead will be called correctly [#2884](https://github.com/openimsdk/open-im-server/pull/2884)
-* fix: webhookBeforeSendSingleMsg will call before black and friend check [#2885](https://github.com/openimsdk/open-im-server/pull/2885)
-
-### Chores
-* chore: remove unused content [#2786](https://github.com/openimsdk/open-im-server/pull/2786)
-
-### Builds
-* build: improve workflows logic. [#2801](https://github.com/openimsdk/open-im-server/pull/2801)
-* build: implement version file update when release. [#2826](https://github.com/openimsdk/open-im-server/pull/2826)
-* build: update mongo and kafka start logic. [#2858](https://github.com/openimsdk/open-im-server/pull/2858)
-* build: create changelog tool and workflows. [#2869](https://github.com/openimsdk/open-im-server/pull/2869)
-* build(deps): bump github.com/golang-jwt/jwt/v4 from 4.5.0 to 4.5.1 [#2851](https://github.com/openimsdk/open-im-server/pull/2851)
-
-### Others
-* Revert: Change group member roleLevel can`t send notification [#2789](https://github.com/openimsdk/open-im-server/pull/2789)
-* Introducing OpenIM Guru on Gurubase.io [#2788](https://github.com/openimsdk/open-im-server/pull/2788)
-* revert: write msg to redis [#2883](https://github.com/openimsdk/open-im-server/pull/2883)
-* @lkzz made their first contribution in https://github.com/openimsdk/open-im-server/pull/2724
- [#2724](https://github.com/openimsdk/open-im-server/pull/2724)
-* @alilestera made their first contribution in https://github.com/openimsdk/open-im-server/pull/2773
- [#2773](https://github.com/openimsdk/open-im-server/pull/2773)
-* @kursataktas made their first contribution in https://github.com/openimsdk/open-im-server/pull/2788
- [#2788](https://github.com/openimsdk/open-im-server/pull/2788)
-* @yoyo930021 made their first contribution in https://github.com/openimsdk/open-im-server/pull/2844
- [#2844](https://github.com/openimsdk/open-im-server/pull/2844)
-* @wikylyu made their first contribution in https://github.com/openimsdk/open-im-server/pull/2861
- [#2861](https://github.com/openimsdk/open-im-server/pull/2861)

CHANGELOG/README.md

@@ -1,4 +0,0 @@
-# CHANGELOGs
-
-- [CHANGELOG-3.8.md](./CHANGELOG-3.8.md)
-

Dockerfile

@@ -1,5 +1,5 @@
-# Use Go 1.22 Alpine as the base image for building the application
+# Use Go 1.21 Alpine as the base image for building the application
-FROM golang:1.22-alpine AS builder
+FROM golang:1.21-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.22-alpine
+FROM golang:1.21-alpine
 
 # Install necessary packages, such as bash
 RUN apk add --no-cache bash

README.md

@@ -16,7 +16,6 @@
 [![Best Practices](https://img.shields.io/badge/Best%20Practices-purple?style=for-the-badge)](https://www.bestpractices.dev/projects/8045)
 [![Good First Issues](https://img.shields.io/github/issues/openimsdk/open-im-server/good%20first%20issue?style=for-the-badge&logo=github)](https://github.com/openimsdk/open-im-server/issues?q=is%3Aissue+is%3Aopen+sort%3Aupdated-desc+label%3A%22good+first+issue%22)
 [![Language](https://img.shields.io/badge/Language-Go-blue.svg?style=for-the-badge&logo=go&logoColor=white)](https://golang.org/)
-[![Gurubase](https://img.shields.io/badge/Gurubase-Ask%20OpenIM%20Guru-006BFF?style=for-the-badge)](https://gurubase.io/g/openim)
 
      
 <p align="center">

cmd/openim-api/main.go

@@ -15,9 +15,10 @@
 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() {

config/mongodb.yml

@@ -8,8 +8,6 @@ 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

config/openim-msggateway.yml

@@ -22,3 +22,5 @@ longConnSvr:
   websocketMaxMsgLen: 4096
   # WebSocket connection handshake timeout in seconds
   websocketTimeout: 10
+
+

config/openim-push.yml

@@ -14,28 +14,28 @@ prometheus:
 
 maxConcurrentWorkers: 3
 #Use geTui for offline push notifications, or choose fcm or jpns; corresponding configuration settings must be specified.
-enable:
+enable: geTui
 geTui:
   pushUrl: https://restapi.getui.com/v2/$appId
-  masterSecret:
+  masterSecret: 
-  appKey:
+  appKey: 
-  intent:
+  intent: 
-  channelID:
+  channelID: 
-  channelName:
+  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.
-jpush:
+jpns:
-  appKey:
+  appKey: 
-  masterSecret:
+  masterSecret: 
-  pushURL:
+  pushURL: 
-  pushIntent:
+  pushIntent: 
 
 # iOS system push sound and badge count
 iosPush:
-  pushSound: xxx
+      pushSound: xxx
-  badgeCount: true
+      badgeCount: true
-  production: false
+      production: false
 
 fullUserCache: true

config/share.yml

@@ -15,4 +15,15 @@ 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
+  maxNumOneEnd: 30
+  customizeLoginNum:
+    ios: 1
+    android: 1
+    windows: 1
+    osx: 1
+    web: 1
+    miniWeb: 1
+    linux: 1
+    aPad: 1
+    iPad: 1
+    admin: 1

deployments/templates/config.yaml

@@ -240,11 +240,11 @@ push:
     channelName: ${GETUI_CHANNEL_NAME}
   fcm:
     serviceAccount: "${FCM_SERVICE_ACCOUNT}"
-  jpush:
+  jpns:
-    appKey: ${JPUSH_APP_KEY}
+    appKey: ${JPNS_APP_KEY}
-    masterSecret: ${JPUSH_MASTER_SECRET}
+    masterSecret: ${JPNS_MASTER_SECRET}
-    pushUrl: ${JPUSH_PUSH_URL}
+    pushUrl: ${JPNS_PUSH_URL}
-    pushIntent: ${JPUSH_PUSH_INTENT}
+    pushIntent: ${JPNS_PUSH_INTENT}
 
 # App manager configuration
 #

docker-compose.yml

@@ -8,35 +8,12 @@ services:
     ports:
       - "37017:27017"
     container_name: mongo
-    command: >
+    command: ["/bin/bash", "-c", "/docker-entrypoint-initdb.d/mongo-init.sh; docker-entrypoint.sh mongod --wiredTigerCacheSizeGB 1 --auth"]
-      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
@@ -94,7 +71,10 @@ 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
@@ -105,11 +85,10 @@ 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:
@@ -145,7 +124,7 @@ services:
       - "11002:80"
     networks:
       - openim
-      
+
 #  prometheus:
 #    image: ${PROMETHEUS_IMAGE}
 #    container_name: prometheus
@@ -192,3 +171,4 @@ services:
 #      - ${DATA_DIR:-./}/components/grafana:/var/lib/grafana
 #    networks:
 #      - openim
+

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              |
-| JPUSH_APP_KEY            | [User Defined]    | JPUSH Application Key             |
+| JPNS_APP_KEY            | [User Defined]    | JPNS Application Key             |
-| JPUSH_MASTER_SECRET      | [User Defined]    | JPUSH Master Secret               |
+| JPNS_MASTER_SECRET      | [User Defined]    | JPNS Master Secret               |
-| JPUSH_PUSH_URL           | [User Defined]    | JPUSH Push Notification URL       |
+| JPNS_PUSH_URL           | [User Defined]    | JPNS Push Notification URL       |
-| JPUSH_PUSH_INTENT        | [User Defined]    | JPUSH Push Intent                 |
+| JPNS_PUSH_INTENT        | [User Defined]    | JPNS Push Intent                 |
 | IM_ADMIN_USERID         | "imAdmin"         | IM Administrator ID              |
 | IM_ADMIN_NAME           | "imAdmin"         | IM Administrator Nickname        |
 | MULTILOGIN_POLICY       | "1"               | Multi-login Policy               |

go.mod

@@ -1,8 +1,6 @@
 module github.com/openimsdk/open-im-server/v3
 
-go 1.22.0
+go 1.21.2
-
-toolchain go1.23.2
 
 require (
 	firebase.google.com/go/v4 v4.14.1
@@ -10,12 +8,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.1
+	github.com/golang-jwt/jwt/v4 v4.5.0
 	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-alpha.55
+	github.com/openimsdk/protocol v0.0.72-pre-release-v3.8.2-alpha.5
-	github.com/openimsdk/tools v0.0.50-alpha.32
+	github.com/openimsdk/tools v0.0.50-alpha.16
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0
@@ -94,7 +92,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.2 // indirect
+	github.com/go-logr/logr v1.4.1 // 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

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.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
+github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
-github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/logr v1.4.1/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.1 h1:JdqV9zKUdtaa9gdPlywC3aeoEsR681PlKC+4F5gQgeo=
+github.com/golang-jwt/jwt/v4 v4.5.0 h1:7cYmW1XlMY7h7ii7UhUyChSgS5wUJEnm9uZVTGqOWzg=
-github.com/golang-jwt/jwt/v4 v4.5.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
+github.com/golang-jwt/jwt/v4 v4.5.0/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-alpha.55 h1:9PPWPHvkFk3neBSbNr+IoOdKIFjxTvEqUfMK/TEq1+8=
+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-alpha.55/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+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.32 h1:JEsUFHFnaYg230TG+Ke3SUnaA2h44t4kABAzEdv5VZw=
+github.com/openimsdk/tools v0.0.50-alpha.16 h1:bC1AQvJMuOHtZm8LZRvN8L5mH1Ws2VYdL+TLTs1iGSc=
-github.com/openimsdk/tools v0.0.50-alpha.32/go.mod h1:r5U6RbxcR4xhKb2fhTmKGC9Yt5LcErHBVt3lhXQIHSo=
+github.com/openimsdk/tools v0.0.50-alpha.16/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4=
 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.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
-github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 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=

internal/api/jssdk/jssdk.go

@@ -1,15 +1,11 @@
 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/protocol/user"
+	"github.com/openimsdk/tools/a2r"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
 	"sort"
@@ -20,22 +16,16 @@ const (
 	defaultGetActiveConversation = 100
 )
 
-func NewJSSdkApi(user user.UserClient, friend relation.FriendClient, group group.GroupClient, msg msg.MsgClient, conv conversation.ConversationClient) *JSSdk {
+func NewJSSdkApi(msg msg.MsgClient, conv conversation.ConversationClient) *JSSdk {
 	return &JSSdk{
-		user:   user,
+		msg:  msg,
-		friend: friend,
+		conv: conv,
-		group:  group,
-		msg:    msg,
-		conv:   conv,
 	}
 }
 
 type JSSdk struct {
-	user   user.UserClient
+	msg  msg.MsgClient
-	friend relation.FriendClient
+	conv conversation.ConversationClient
-	group  group.GroupClient
-	msg    msg.MsgClient
-	conv   conversation.ConversationClient
 }
 
 func (x *JSSdk) GetActiveConversations(c *gin.Context) {
@@ -46,71 +36,25 @@ func (x *JSSdk) GetConversations(c *gin.Context) {
 	call(c, x.getConversations)
 }
 
-func (x *JSSdk) fillConversations(ctx context.Context, conversations []*jssdk.ConversationMsg) error {
+func (x *JSSdk) getActiveConversations(ctx *gin.Context) (*ConversationsResp, error) {
-	if len(conversations) == 0 {
+	req, err := a2r.ParseRequest[ActiveConversationsReq](ctx)
-		return nil
+	if err != nil {
+		return nil, err
 	}
-	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
 	}
-	req.OwnerUserID = mcontext.GetOpUserID(ctx)
+	opUserID := mcontext.GetOpUserID(ctx)
 	conversationIDs, err := field(ctx, x.conv.GetConversationIDs,
-		&conversation.GetConversationIDsReq{UserID: req.OwnerUserID}, (*conversation.GetConversationIDsResp).GetConversationIDs)
+		&conversation.GetConversationIDsReq{UserID: opUserID}, (*conversation.GetConversationIDsResp).GetConversationIDs)
 	if err != nil {
 		return nil, err
 	}
 	if len(conversationIDs) == 0 {
-		return &jssdk.GetActiveConversationsResp{}, nil
+		return &ConversationsResp{}, nil
 	}
 	readSeq, err := field(ctx, x.msg.GetHasReadSeqs,
-		&msg.GetHasReadSeqsReq{UserID: req.OwnerUserID, ConversationIDs: conversationIDs}, (*msg.SeqsInfoResp).GetMaxSeqs)
+		&msg.GetHasReadSeqsReq{UserID: opUserID, ConversationIDs: conversationIDs}, (*msg.SeqsInfoResp).GetMaxSeqs)
 	if err != nil {
 		return nil, err
 	}
@@ -120,24 +64,24 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 		return nil, err
 	}
 	if len(activeConversation) == 0 {
-		return &jssdk.GetActiveConversationsResp{}, nil
+		return &ConversationsResp{}, nil
 	}
 	sortConversations := sortActiveConversations{
 		Conversation: activeConversation,
 	}
 	if len(activeConversation) > 1 {
 		pinnedConversationIDs, err := field(ctx, x.conv.GetPinnedConversationIDs,
-			&conversation.GetPinnedConversationIDsReq{UserID: req.OwnerUserID}, (*conversation.GetPinnedConversationIDsResp).GetConversationIDs)
+			&conversation.GetPinnedConversationIDsReq{UserID: opUserID}, (*conversation.GetPinnedConversationIDsResp).GetConversationIDs)
 		if err != nil {
 			return nil, err
 		}
 		sortConversations.PinnedConversationIDs = datautil.SliceSet(pinnedConversationIDs)
 	}
 	sort.Sort(&sortConversations)
-	sortList := sortConversations.Top(int(req.Count))
+	sortList := sortConversations.Top(req.Count)
 	conversations, err := field(ctx, x.conv.GetConversations,
 		&conversation.GetConversationsReq{
-			OwnerUserID: req.OwnerUserID,
+			OwnerUserID: opUserID,
 			ConversationIDs: datautil.Slice(sortList, func(c *msg.ActiveConversation) string {
 				return c.ConversationID
 			})}, (*conversation.GetConversationsResp).GetConversations)
@@ -146,7 +90,7 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 	}
 	msgs, err := field(ctx, x.msg.GetSeqMessage,
 		&msg.GetSeqMessageReq{
-			UserID: req.OwnerUserID,
+			UserID: opUserID,
 			Conversations: datautil.Slice(sortList, func(c *msg.ActiveConversation) *msg.ConversationSeqs {
 				return &msg.ConversationSeqs{
 					ConversationID: c.ConversationID,
@@ -160,7 +104,7 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 	conversationMap := datautil.SliceToMap(conversations, func(c *conversation.Conversation) string {
 		return c.ConversationID
 	})
-	resp := make([]*jssdk.ConversationMsg, 0, len(sortList))
+	resp := make([]ConversationMsg, 0, len(sortList))
 	for _, c := range sortList {
 		conv, ok := conversationMap[c.ConversationID]
 		if !ok {
@@ -170,16 +114,13 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 		if msgList, ok := msgs[c.ConversationID]; ok && len(msgList.Msgs) > 0 {
 			lastMsg = msgList.Msgs[0]
 		}
-		resp = append(resp, &jssdk.ConversationMsg{
+		resp = append(resp, 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]
@@ -187,20 +128,24 @@ func (x *JSSdk) getActiveConversations(ctx context.Context, req *jssdk.GetActive
 			unreadCount += count
 		}
 	}
-	return &jssdk.GetActiveConversationsResp{
+	return &ConversationsResp{
 		Conversations: resp,
 		UnreadCount:   unreadCount,
 	}, nil
 }
 
-func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversationsReq) (*jssdk.GetConversationsResp, error) {
+func (x *JSSdk) getConversations(ctx *gin.Context) (*ConversationsResp, error) {
+	req, err := a2r.ParseRequest[conversation.GetConversationsReq](ctx)
+	if err != nil {
+		return nil, err
+	}
 	req.OwnerUserID = mcontext.GetOpUserID(ctx)
-	conversations, err := field(ctx, x.conv.GetConversations, &conversation.GetConversationsReq{OwnerUserID: req.OwnerUserID, ConversationIDs: req.ConversationIDs}, (*conversation.GetConversationsResp).GetConversations)
+	conversations, err := field(ctx, x.conv.GetConversations, req, (*conversation.GetConversationsResp).GetConversations)
 	if err != nil {
 		return nil, err
 	}
 	if len(conversations) == 0 {
-		return &jssdk.GetConversationsResp{}, nil
+		return &ConversationsResp{}, nil
 	}
 	req.ConversationIDs = datautil.Slice(conversations, func(c *conversation.Conversation) string {
 		return c.ConversationID
@@ -232,22 +177,19 @@ func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversation
 			return nil, err
 		}
 	}
-	resp := make([]*jssdk.ConversationMsg, 0, len(conversations))
+	resp := make([]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, &jssdk.ConversationMsg{
+		resp = append(resp, 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]
@@ -255,7 +197,7 @@ func (x *JSSdk) getConversations(ctx context.Context, req *jssdk.GetConversation
 			unreadCount += count
 		}
 	}
-	return &jssdk.GetConversationsResp{
+	return &ConversationsResp{
 		Conversations: resp,
 		UnreadCount:   unreadCount,
 	}, nil

internal/api/jssdk/stu.go

@@ -0,0 +1,22 @@
+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"`
+}

internal/api/jssdk/tools.go

@@ -3,14 +3,8 @@ 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) {
@@ -22,56 +16,11 @@ func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A
 	return get(resp), nil
 }
 
-func call[A, B any](c *gin.Context, fn func(ctx context.Context, req *A) (*B, error)) {
+func call[R any](c *gin.Context, fn func(ctx *gin.Context) (R, error)) {
-	var isJSON bool
+	resp, err := fn(c)
-	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)
-		apiresp.GinSuccess(c, resp)
-		return
-	}
-	body, err := proto.Marshal(any(resp).(proto.Message))
-	if err != nil {
-		apiresp.GinError(c, err)
-		return
-	}
-	apiresp.GinSuccess(c, body)
 }

internal/api/jssdk_test.go

@@ -0,0 +1,37 @@
+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)
+
+}

internal/api/msg.go

@@ -173,8 +173,6 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
 		data = apistruct.AtElem{}
 	case constant.Custom:
 		data = apistruct.CustomElem{}
-	case constant.Stream:
-		data = apistruct.StreamMsgElem{}
 	case constant.OANotification:
 		data = apistruct.OANotificationElem{}
 		req.SessionType = constant.NotificationChatType
@@ -375,11 +373,3 @@ func (m *MessageApi) SearchMsg(c *gin.Context) {
 func (m *MessageApi) GetServerTime(c *gin.Context) {
 	a2r.Call(msg.MsgClient.GetServerTime, m.Client, c)
 }
-
-func (m *MessageApi) GetStreamMsg(c *gin.Context) {
-	a2r.Call(msg.MsgClient.GetStreamMsg, m.Client, c)
-}
-
-func (m *MessageApi) AppendStreamMsg(c *gin.Context) {
-	a2r.Call(msg.MsgClient.AppendStreamMsg, m.Client, c)
-}

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.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc))
+	r.Use(prommetricsGin(), gin.Recovery(), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc))
 	u := NewUserApi(*userRpc)
 	m := NewMessageApi(messageRpc, userRpc, config.Share.IMAdminUserID)
-	j := jssdk.NewJSSdkApi(userRpc.Client, friendRpc.Client, groupRpc.Client, messageRpc.Client, conversationRpc.Client)
+	j := jssdk.NewJSSdkApi(messageRpc.Client, conversationRpc.Client)
 	userRouterGroup := r.Group("/user")
 	{
 		userRouterGroup.POST("/user_register", u.UserRegister)
@@ -222,8 +222,6 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
 		msgGroup.POST("/batch_send_msg", m.BatchSendMsg)
 		msgGroup.POST("/check_msg_is_send_success", m.CheckMsgIsSendSuccess)
 		msgGroup.POST("/get_server_time", m.GetServerTime)
-		msgGroup.POST("/get_stream_msg", m.GetStreamMsg)
-		msgGroup.POST("/append_stream_msg", m.AppendStreamMsg)
 	}
 	// Conversation
 	conversationGroup := r.Group("/conversation")

internal/msggateway/client.go

@@ -16,9 +16,7 @@ package msggateway
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"github.com/openimsdk/tools/mw"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
@@ -71,8 +69,6 @@ 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
@@ -98,17 +94,11 @@ 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{})
 }
 
@@ -169,12 +159,9 @@ func (c *Client) readMessage() {
 				return
 			}
 		case MessageText:
-			_ = c.conn.SetReadDeadline(pongWait)
+			c.closedErr = ErrNotSupportMessageProtocol
-			parseDataErr := c.handlerTextMessage(message)
+			return
-			if parseDataErr != nil {
+
-				c.closedErr = parseDataErr
-				return
-			}
 		case PingMessage:
 			err := c.writePongMsg("")
 			log.ZError(c.ctx, "writePongMsg", err)
@@ -201,7 +188,7 @@ func (c *Client) handleMessage(message []byte) error {
 	var binaryReq = getReq()
 	defer freeReq(binaryReq)
 
-	err := c.Encoder.Decode(message, binaryReq)
+	err := c.longConnServer.Decode(message, binaryReq)
 	if err != nil {
 		return err
 	}
@@ -348,7 +335,7 @@ func (c *Client) writeBinaryMsg(resp Resp) error {
 		return nil
 	}
 
-	encodedBuf, err := c.Encoder.Encode(resp)
+	encodedBuf, err := c.longConnServer.Encode(resp)
 	if err != nil {
 		return err
 	}
@@ -376,11 +363,6 @@ 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()
@@ -437,28 +419,3 @@ 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)
-	}
-}

internal/msggateway/constant.go

@@ -27,12 +27,6 @@ const (
 	GzipCompressionProtocol = "gzip"
 	BackgroundStatus        = "isBackground"
 	SendResponse            = "isMsgResp"
-	SDKType                 = "sdkType"
-)
-
-const (
-	GoSDK = "go"
-	JsSDK = "js"
 )
 
 const (

internal/msggateway/context.go

@@ -153,14 +153,6 @@ 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 {
@@ -201,11 +193,7 @@ 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
 }

internal/msggateway/encoder.go

@@ -17,7 +17,6 @@ package msggateway
 import (
 	"bytes"
 	"encoding/gob"
-	"encoding/json"
 
 	"github.com/openimsdk/tools/errs"
 )
@@ -29,12 +28,12 @@ type Encoder interface {
 
 type GobEncoder struct{}
 
-func NewGobEncoder() Encoder {
+func NewGobEncoder() *GobEncoder {
-	return GobEncoder{}
+	return &GobEncoder{}
 }
 
-func (g GobEncoder) Encode(data any) ([]byte, error) {
+func (g *GobEncoder) Encode(data any) ([]byte, error) {
-	var buff bytes.Buffer
+	buff := bytes.Buffer{}
 	enc := gob.NewEncoder(&buff)
 	if err := enc.Encode(data); err != nil {
 		return nil, errs.WrapMsg(err, "GobEncoder.Encode failed", "action", "encode")
@@ -42,7 +41,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 {
@@ -50,25 +49,3 @@ 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
-}

internal/msggateway/hub_server.go

@@ -83,11 +83,17 @@ 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")
 	}
@@ -149,7 +155,6 @@ 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 {
@@ -215,7 +220,10 @@ 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 {

internal/msggateway/message_handler.go

@@ -16,7 +16,6 @@ package msggateway
 
 import (
 	"context"
-	"encoding/json"
 	"sync"
 
 	"github.com/go-playground/validator/v10"
@@ -32,16 +31,6 @@ 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"`

internal/msggateway/ws_server.go

@@ -37,6 +37,7 @@ type LongConnServer interface {
 	SetKickHandlerInfo(i *kickHandler)
 	SubUserOnlineStatus(ctx context.Context, client *Client, data *Req) ([]byte, error)
 	Compressor
+	Encoder
 	MessageHandler
 }
 
@@ -60,7 +61,7 @@ type WsServer struct {
 	authClient        *rpcclient.Auth
 	disCov            discovery.SvcDiscoveryRegistry
 	Compressor
-	//Encoder
+	Encoder
 	MessageHandler
 	webhookClient *webhook.Client
 }
@@ -134,6 +135,7 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer {
 		clients:         newUserMap(),
 		subscription:    newSubscription(),
 		Compressor:      NewGzipCompressor(),
+		Encoder:         NewGobEncoder(),
 		webhookClient:   webhook.NewWebhookClient(msgGatewayConfig.WebhooksConfig.URL),
 	}
 }
@@ -276,7 +278,14 @@ 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 {
@@ -318,6 +327,11 @@ 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
@@ -342,7 +356,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 			log.ZWarn(newClient.ctx, "InvalidateToken err", err, "userID", newClient.UserID,
 				"platformID", newClient.PlatformID)
 		}
-	case constant.AllLoginButSameClassKick:
+	case constant.PcMobileAndWeb:
 		clients, ok := ws.clients.GetAll(newClient.UserID)
 		if !ok {
 			return
@@ -356,6 +370,21 @@ 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
 	}
 }
 

internal/msgtransfer/init.go

@@ -136,11 +136,6 @@ 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

internal/msgtransfer/online_history_msg_handler.go

@@ -19,7 +19,6 @@ import (
 	"encoding/json"
 	"errors"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-	"github.com/openimsdk/tools/mw"
 	"strconv"
 	"strings"
 	"sync"
@@ -347,12 +346,6 @@ 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 {

internal/push/offlinepush/dummy/push.go

@@ -29,6 +29,5 @@ 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
 }

internal/push/offlinepush/jpush/body/notification.go

@@ -15,7 +15,6 @@
 package body
 
 import (
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
@@ -27,44 +26,38 @@ 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 map[string]string `json:"extras,omitempty"`
+	Extras Extras `json:"extras"`
 }
 type Ios struct {
-	Alert          IosAlert          `json:"alert,omitempty"`
+	Alert          string `json:"alert,omitempty"`
-	Sound          string            `json:"sound,omitempty"`
+	Sound          string `json:"sound,omitempty"`
-	Badge          string            `json:"badge,omitempty"`
+	Badge          string `json:"badge,omitempty"`
-	Extras         map[string]string `json:"extras,omitempty"`
+	Extras         Extras `json:"extras"`
-	MutableContent bool              `json:"mutable-content"`
+	MutableContent bool   `json:"mutable-content"`
 }
 
-type IosAlert struct {
+type Extras struct {
-	Title string `json:"title,omitempty"`
+	ClientMsgID string `json:"clientMsgID"`
-	Body  string `json:"body,omitempty"`
 }
 
-func (n *Notification) SetAlert(alert string, title string, opts *options.Opts) {
+func (n *Notification) SetAlert(alert string) {
 	n.Alert = alert
 	n.Android.Alert = alert
-	n.Android.Title = title
+	n.IOS.Alert = alert
-	n.IOS.Alert.Body = alert
+	n.IOS.Sound = "default"
-	n.IOS.Alert.Title = title
+	n.IOS.Badge = "+1"
-	n.IOS.Sound = opts.IOSPushSound
-	if opts.IOSBadgeCount {
-		n.IOS.Badge = "+1"
-	}
 }
 
-func (n *Notification) SetExtras(extras map[string]string) {
+func (n *Notification) SetExtras(extras Extras) {
 	n.IOS.Extras = extras
 	n.Android.Extras = extras
 }
 
 func (n *Notification) SetAndroidIntent(pushConf *config.Push) {
-	n.Android.Intent.URL = pushConf.JPush.PushIntent
+	n.Android.Intent.URL = pushConf.JPNS.PushIntent
 }
 
 func (n *Notification) IOSEnableMutableContent() {

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,23 +57,17 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin
 	var au body.Audience
 	au.SetAlias(userIDs)
 	var no body.Notification
-	extras := make(map[string]string)
+	var extras body.Extras
-	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(content, title, opts)
+	no.SetAlert(title)
 	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
@@ -82,26 +76,19 @@ func (j *JPush) Push(ctx context.Context, userIDs []string, title, content strin
 	pushObj.SetNotification(&no)
 	pushObj.SetMessage(&msg)
 	pushObj.SetOptions(&opt)
-	var resp map[string]any
+	var resp any
-	return j.request(ctx, pushObj, &resp, 5)
+	return j.request(ctx, pushObj, resp, 5)
 }
 
-func (j *JPush) request(ctx context.Context, po body.PushObj, resp *map[string]any, timeout int) error {
+func (j *JPush) request(ctx context.Context, po body.PushObj, resp any, timeout int) error {
-	err := j.httpClient.PostReturn(
+	return j.httpClient.PostReturn(
 		ctx,
-		j.pushConf.JPush.PushURL,
+		j.pushConf.JPNS.PushURL,
 		map[string]string{
-			"Authorization": j.getAuthorization(j.pushConf.JPush.AppKey, j.pushConf.JPush.MasterSecret),
+			"Authorization": j.getAuthorization(j.pushConf.JPNS.AppKey, j.pushConf.JPNS.MasterSecret),
 		},
 		po,
 		resp,
 		timeout,
 	)
-	if err != nil {
-		return err
-	}
-	if (*resp)["sendno"] != "0" {
-		return fmt.Errorf("jpush push failed %v", resp)
-	}
-	return nil
 }

internal/push/offlinepush/offlinepusher.go

@@ -23,6 +23,8 @@ 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"
 )
 
@@ -49,6 +51,7 @@ 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
 }

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{ClientMsgID: msg.ClientMsgID}}
+	opts = &options.Opts{Signal: &options.Signal{}}
 	if msg.OfflinePushInfo != nil {
 		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
 		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound

internal/push/push_handler.go

@@ -4,10 +4,6 @@ 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"
@@ -31,6 +27,9 @@ 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 {
@@ -166,21 +165,17 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *
 			return nil
 		}
 	}
-	needOfflinePushUserID := []string{msg.RecvID}
+	offlinePushUserID := []string{msg.RecvID}
-	var offlinePushUserID []string
 
 	//receiver offline push
-	if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserID, msg, &offlinePushUserID); err != nil {
+	if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush,
+		offlinePushUserID, msg, nil); 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, "needOfflinePushUserID", needOfflinePushUserID, "msg", msg)
+		log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePushUserID", offlinePushUserID, "msg", msg)
 		return nil
 	}
 
@@ -340,7 +335,6 @@ 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)
@@ -370,7 +364,7 @@ func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, conten
 		IsAtSelf   bool     `json:"isAtSelf"`
 	}
 
-	opts = &options.Opts{Signal: &options.Signal{ClientMsgID: msg.ClientMsgID}}
+	opts = &options.Opts{Signal: &options.Signal{}}
 	if msg.OfflinePushInfo != nil {
 		opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
 		opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound

internal/rpc/auth/auth.go

@@ -16,7 +16,6 @@ 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"
@@ -67,7 +66,6 @@ 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,
 	})
@@ -131,10 +129,6 @@ 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
@@ -196,7 +190,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID
 	}
 
 	m, err := s.authDatabase.GetTokensWithoutError(ctx, userID, int(platformID))
-	if err != nil && errors.Is(err, redis.Nil) {
+	if err != nil && err != redis.Nil {
 		return err
 	}
 	for k := range m {
@@ -214,7 +208,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 && errors.Is(err, redis.Nil) {
+	if err != nil && err != redis.Nil {
 		return nil, err
 	}
 	if m == nil {

internal/rpc/conversation/conversation.go

@@ -261,35 +261,27 @@ 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
 		}
 	}

internal/rpc/group/group.go

@@ -1811,6 +1811,7 @@ 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
@@ -1819,11 +1820,10 @@ func (g *groupServer) GetSpecifiedUserGroupRequestInfo(ctx context.Context, req
 		adminIDs = append(adminIDs, owners[0].UserID)
 		adminIDs = append(adminIDs, g.config.Share.IMAdminUserID...)
 
-		if !datautil.Contain(opUserID, adminIDs...) {
+		if !datautil.Contain(req.UserID, 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

internal/rpc/msg/as_read.go

@@ -16,7 +16,6 @@ package msg
 
 import (
 	"context"
-	"errors"
 
 	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 	"github.com/openimsdk/protocol/constant"
@@ -109,7 +108,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 && !errors.Is(err, redis.Nil) {
+	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return nil, err
 	}
 	if hasReadSeq > currentHasReadSeq {
@@ -137,7 +136,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 && errors.Is(err, redis.Nil) {
+	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return nil, err
 	}
 	var seqs []int64
@@ -181,23 +180,14 @@ func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkCon
 			req.UserID, seqs, hasReadSeq)
 	}
 
-	if conversation.ConversationType == constant.SingleChatType {
+	reqCall := &cbapi.CallbackGroupMsgReadReq{
-		reqCall := &cbapi.CallbackSingleMsgReadReq{
+		SendID:       conversation.OwnerUserID,
-			ConversationID: conversation.ConversationID,
+		ReceiveID:    req.UserID,
-			UserID:         conversation.OwnerUserID,
+		UnreadMsgNum: req.HasReadSeq,
-			Seqs:           req.Seqs,
+		ContentType:  int64(conversation.ConversationType),
-			ContentType:    conversation.ConversationType,
-		}
-		m.webhookAfterSingleMsgRead(ctx, &m.config.WebhooksConfig.AfterSingleMsgRead, reqCall)
-	} else if conversation.ConversationType == constant.ReadGroupChatType {
-		reqCall := &cbapi.CallbackGroupMsgReadReq{
-			SendID:       conversation.OwnerUserID,
-			ReceiveID:    req.UserID,
-			UnreadMsgNum: req.HasReadSeq,
-			ContentType:  int64(conversation.ConversationType),
-		}
-		m.webhookAfterGroupMsgRead(ctx, &m.config.WebhooksConfig.AfterGroupMsgRead, reqCall)
 	}
+
+	m.webhookAfterGroupMsgRead(ctx, &m.config.WebhooksConfig.AfterGroupMsgRead, reqCall)
 	return &msg.MarkConversationAsReadResp{}, nil
 }
 

internal/rpc/msg/notification.go

@@ -48,7 +48,3 @@ func (m *MsgNotificationSender) MarkAsReadNotification(ctx context.Context, conv
 	}
 	m.NotificationWithSessionType(ctx, sendID, recvID, constant.HasReadReceipt, sessionType, tips)
 }
-
-func (m *MsgNotificationSender) StreamMsgNotification(ctx context.Context, sendID string, recvID string, sessionType int32, tips *sdkws.StreamMsgTips) {
-	m.NotificationWithSessionType(ctx, sendID, recvID, constant.StreamMsgNotification, sessionType, tips)
-}

internal/rpc/msg/send.go

@@ -16,7 +16,6 @@ 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"
@@ -35,11 +34,6 @@ import (
 func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
 	if req.MsgData != nil {
 		m.encapsulateMsgData(req.MsgData)
-		if req.MsgData.ContentType == constant.Stream {
-			if err := m.handlerStreamMsg(ctx, req.MsgData); err != nil {
-				return nil, err
-			}
-		}
 		switch req.MsgData.SessionType {
 		case constant.SingleChatType:
 			return m.sendMsgSingleChat(ctx, req)
@@ -84,15 +78,8 @@ 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
@@ -179,6 +166,9 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 		prommetrics.SingleChatMsgProcessFailedCounter.Inc()
 		return nil, nil
 	} else {
+		if err = m.webhookBeforeSendSingleMsg(ctx, &m.config.WebhooksConfig.BeforeSendSingleMsg, req); err != nil {
+			return nil, err
+		}
 		if err := m.webhookBeforeMsgModify(ctx, &m.config.WebhooksConfig.BeforeMsgModify, req); err != nil {
 			return nil, err
 		}

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 && !errors.Is(err, redis.Nil) {
+	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return nil, err
 	}
 	return &pbmsg.GetConversationMaxSeqResp{MaxSeq: maxSeq}, nil

internal/rpc/msg/server.go

@@ -42,9 +42,8 @@ type (
 
 	// MsgServer encapsulates dependencies required for message handling.
 	msgServer struct {
-		RegisterCenter         discovery.SvcDiscoveryRegistry // Service discovery registry for service registration.
+		RegisterCenter         discovery.SvcDiscoveryRegistry   // Service discovery registry for service registration.
-		MsgDatabase            controller.CommonMsgDatabase   // Interface for message database operations.
+		MsgDatabase            controller.CommonMsgDatabase     // Interface for message database operations.
-		StreamMsgDatabase      controller.StreamMsgDatabase
 		Conversation           *rpcclient.ConversationRpcClient // RPC client for conversation service.
 		UserLocalCache         *rpccache.UserLocalCache         // Local cache for user data.
 		FriendLocalCache       *rpccache.FriendLocalCache       // Local cache for friend data.
@@ -102,10 +101,6 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	streamMsg, err := mgo.NewStreamMsgMongo(mgocli.GetDB())
-	if err != nil {
-		return err
-	}
 	seqUserCache := redis.NewSeqUserCacheRedis(rdb, seqUser)
 	msgDatabase, err := controller.NewCommonMsgDatabase(msgDocModel, msgModel, seqUserCache, seqConversationCache, &config.KafkaConfig)
 	if err != nil {
@@ -114,7 +109,6 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	s := &msgServer{
 		Conversation:           &conversationClient,
 		MsgDatabase:            msgDatabase,
-		StreamMsgDatabase:      controller.NewStreamMsgDatabase(streamMsg),
 		RegisterCenter:         client,
 		UserLocalCache:         rpccache.NewUserLocalCache(userRpcClient, &config.LocalCacheConfig, rdb),
 		GroupLocalCache:        rpccache.NewGroupLocalCache(groupRpcClient, &config.LocalCacheConfig, rdb),

internal/rpc/msg/stream_msg.go

@@ -1,114 +0,0 @@
-package msg
-
-import (
-	"context"
-	"fmt"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
-	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/protocol/msg"
-	"github.com/openimsdk/protocol/sdkws"
-	"github.com/openimsdk/tools/errs"
-	"time"
-)
-
-const StreamDeadlineTime = time.Second * 60 * 10
-
-func (m *msgServer) handlerStreamMsg(ctx context.Context, msgData *sdkws.MsgData) error {
-	now := time.Now()
-	val := &model.StreamMsg{
-		ClientMsgID:    msgData.ClientMsgID,
-		ConversationID: msgprocessor.GetConversationIDByMsg(msgData),
-		UserID:         msgData.SendID,
-		CreateTime:     now,
-		DeadlineTime:   now.Add(StreamDeadlineTime),
-	}
-	return m.StreamMsgDatabase.CreateStreamMsg(ctx, val)
-}
-
-func (m *msgServer) getStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
-	res, err := m.StreamMsgDatabase.GetStreamMsg(ctx, clientMsgID)
-	if err != nil {
-		return nil, err
-	}
-	now := time.Now()
-	if !res.End && res.DeadlineTime.Before(now) {
-		res.End = true
-		res.DeadlineTime = now
-		_ = m.StreamMsgDatabase.AppendStreamMsg(ctx, res.ClientMsgID, 0, nil, true, now)
-	}
-	return res, nil
-}
-
-func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) {
-	res, err := m.getStreamMsg(ctx, req.ClientMsgID)
-	if err != nil {
-		return nil, err
-	}
-	if res.End {
-		return nil, errs.ErrNoPermission.WrapMsg("stream msg is end")
-	}
-	if len(res.Packets) < int(req.StartIndex) {
-		return nil, errs.ErrNoPermission.WrapMsg("start index is invalid")
-	}
-	if val := len(res.Packets) - int(req.StartIndex); val > 0 {
-		exist := res.Packets[int(req.StartIndex):]
-		for i, s := range exist {
-			if len(req.Packets) == 0 {
-				break
-			}
-			if s != req.Packets[i] {
-				return nil, errs.ErrNoPermission.WrapMsg(fmt.Sprintf("packet %d has been written and is inconsistent", i))
-			}
-			req.StartIndex++
-			req.Packets = req.Packets[1:]
-		}
-	}
-	if len(req.Packets) == 0 && res.End == req.End {
-		return &msg.AppendStreamMsgResp{}, nil
-	}
-	deadlineTime := time.Now().Add(StreamDeadlineTime)
-	if err := m.StreamMsgDatabase.AppendStreamMsg(ctx, req.ClientMsgID, int(req.StartIndex), req.Packets, req.End, deadlineTime); err != nil {
-		return nil, err
-	}
-	conversation, err := m.Conversation.GetConversation(ctx, res.UserID, res.ConversationID)
-	if err != nil {
-		return nil, err
-	}
-	tips := &sdkws.StreamMsgTips{
-		ConversationID: res.ConversationID,
-		ClientMsgID:    res.ClientMsgID,
-		StartIndex:     req.StartIndex,
-		Packets:        req.Packets,
-		End:            req.End,
-	}
-	var (
-		recvID      string
-		sessionType int32
-	)
-	if conversation.GroupID == "" {
-		sessionType = constant.SingleChatType
-		recvID = conversation.UserID
-	} else {
-		sessionType = constant.ReadGroupChatType
-		recvID = conversation.GroupID
-	}
-	m.msgNotificationSender.StreamMsgNotification(ctx, res.UserID, recvID, sessionType, tips)
-	return &msg.AppendStreamMsgResp{}, nil
-}
-
-func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) {
-	res, err := m.getStreamMsg(ctx, req.ClientMsgID)
-	if err != nil {
-		return nil, err
-	}
-	return &msg.GetStreamMsgResp{
-		ClientMsgID:    res.ClientMsgID,
-		ConversationID: res.ConversationID,
-		UserID:         res.UserID,
-		Packets:        res.Packets,
-		End:            res.End,
-		CreateTime:     res.CreateTime.UnixMilli(),
-		DeadlineTime:   res.DeadlineTime.UnixMilli(),
-	}, nil
-}

internal/rpc/msg/verify.go

@@ -59,9 +59,6 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
 			data.MsgData.ContentType >= constant.NotificationBegin {
 			return nil
 		}
-		if err := m.webhookBeforeSendSingleMsg(ctx, &m.config.WebhooksConfig.BeforeSendSingleMsg, data); err != nil {
-			return err
-		}
 		black, err := m.FriendLocalCache.IsBlack(ctx, data.MsgData.SendID, data.MsgData.RecvID)
 		if err != nil {
 			return err

internal/rpc/relation/friend.go

@@ -273,14 +273,7 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *relation.SetFri
 	return &relation.SetFriendRemarkResp{}, nil
 }
 
-func (s *friendServer) GetFriendInfo(ctx context.Context, req *relation.GetFriendInfoReq) (*relation.GetFriendInfoResp, error) {
+// ok.
-	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) {

internal/rpc/third/third.go

@@ -74,7 +74,6 @@ 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 (

internal/rpc/third/tool.go

@@ -82,11 +82,3 @@ 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
-}

pkg/apistruct/msg.go

@@ -81,11 +81,6 @@ type TextElem struct {
 	Content string `json:"content" validate:"required"`
 }
 
-type StreamMsgElem struct {
-	Type    string `mapstructure:"type" validate:"required"`
-	Content string `mapstructure:"content" validate:"required"`
-}
-
 type RevokeElem struct {
 	RevokeMsgClientID string `mapstructure:"revokeMsgClientID" validate:"required"`
 }

pkg/common/config/config.go

@@ -69,7 +69,6 @@ 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"`
 }
@@ -213,12 +212,12 @@ type Push struct {
 		FilePath string `mapstructure:"filePath"`
 		AuthURL  string `mapstructure:"authURL"`
 	} `mapstructure:"fcm"`
-	JPush struct {
+	JPNS struct {
 		AppKey       string `mapstructure:"appKey"`
 		MasterSecret string `mapstructure:"masterSecret"`
 		PushURL      string `mapstructure:"pushURL"`
 		PushIntent   string `mapstructure:"pushIntent"`
-	} `mapstructure:"jpush"`
+	} `mapstructure:"jpns"`
 	IOSPush struct {
 		PushSound  string `mapstructure:"pushSound"`
 		BadgeCount bool   `mapstructure:"badgeCount"`
@@ -369,8 +368,20 @@ type Share struct {
 }
 
 type MultiLogin struct {
-	Policy       int `mapstructure:"policy"`
+	Policy            int `mapstructure:"policy"`
-	MaxNumOneEnd int `mapstructure:"maxNumOneEnd"`
+	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"`
 }
 
 type RpcRegisterName struct {
@@ -479,7 +490,6 @@ func (m *Mongo) Build() *mongoutil.Config {
 		Database:    m.Database,
 		Username:    m.Username,
 		Password:    m.Password,
-		AuthSource:  m.AuthSource,
 		MaxPoolSize: m.MaxPoolSize,
 		MaxRetry:    m.MaxRetry,
 	}

pkg/common/convert/friend.go

@@ -18,7 +18,6 @@ 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"
@@ -36,7 +35,9 @@ 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
@@ -52,7 +53,11 @@ func FriendDB2Pb(ctx context.Context, friendDB *model.Friend, getUsers func(ctx
 	}, 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
 	}
@@ -81,21 +86,7 @@ func FriendsDB2Pb(ctx context.Context, friendsDB []*model.Friend, getUsers func(
 		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) {

pkg/common/storage/cache/redis/conversation.go

@@ -38,7 +38,7 @@ const (
 func NewConversationRedis(rdb redis.UniversalClient, localCache *config.LocalCache, opts *rockscache.Options, db database.Conversation) cache.ConversationCache {
 	batchHandler := NewBatchDeleterRedis(rdb, opts, []string{localCache.Conversation.Topic})
 	c := localCache.Conversation
-	log.ZDebug(context.Background(), "conversation local cache init", "Topic", c.Topic, "SlotNum", c.SlotNum, "SlotSize", c.SlotSize, "enable", c.Enable())
+	log.ZDebug(context.Background(), "black local cache init", "Topic", c.Topic, "SlotNum", c.SlotNum, "SlotSize", c.SlotSize, "enable", c.Enable())
 	return &ConversationRedisCache{
 		BatchDeleter:   batchHandler,
 		rcClient:       rockscache.NewClient(rdb, *opts),

pkg/common/storage/cache/redis/msg.go

@@ -1,3 +1,17 @@
+// 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 (

pkg/common/storage/controller/auth.go

@@ -2,14 +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"
 )
 
@@ -25,8 +26,9 @@ type AuthDatabase interface {
 }
 
 type multiLoginConfig struct {
-	Policy       int
+	Policy            int
-	MaxNumOneEnd int
+	MaxNumOneEnd      int
+	CustomizeLoginNum map[int]int
 }
 
 type authDatabase struct {
@@ -34,16 +36,25 @@ type authDatabase struct {
 	accessSecret string
 	accessExpire int64
 	multiLogin   multiLoginConfig
-	adminUserIDs []string
 }
 
-func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin, adminUserIDs []string) AuthDatabase {
+func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin) AuthDatabase {
 	return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLogin: multiLoginConfig{
 		Policy:       multiLogin.Policy,
 		MaxNumOneEnd: multiLogin.MaxNumOneEnd,
-	},
+		CustomizeLoginNum: map[int]int{
-		adminUserIDs: adminUserIDs,
+			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,
+		},
+	}}
 }
 
 // If the result is empty.
@@ -80,31 +91,27 @@ func (a *authDatabase) BatchSetTokenMapByUidPid(ctx context.Context, tokens []st
 
 // Create Token.
 func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) {
-	isAdmin := authverify.IsManagerUserID(userID, a.adminUserIDs)
+	tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID)
-	if !isAdmin {
+	if err != nil {
-		tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID)
+		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)
 		if err != nil {
 			return "", err
 		}
-
+	}
-		deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
+	if len(kickedTokenKey) != 0 {
-		if err != nil {
+		for _, k := range kickedTokenKey {
-			return "", err
+			err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
-		}
-		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)
-			}
 		}
 	}
 
@@ -115,12 +122,9 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
 		return "", errs.WrapMsg(err, "token.SignedString")
 	}
 
-	if !isAdmin {
+	if err = a.cache.SetTokenFlagEx(ctx, userID, platformID, tokenString, constant.NormalToken); err != nil {
-		if err = a.cache.SetTokenFlagEx(ctx, userID, platformID, tokenString, constant.NormalToken); err != nil {
+		return "", err
-			return "", err
-		}
 	}
-
 	return tokenString, nil
 }
 
@@ -168,8 +172,17 @@ 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:
-		unkickTerminal = constant.TerminalPC
+		if unkickTerminal == "" {
+			unkickTerminal = constant.TerminalPC
+		}
 		if constant.PlatformIDToClass(platformID) != unkickTerminal {
 			for plt, ts := range loginTokenMap {
 				if constant.PlatformIDToClass(plt) != unkickTerminal {
@@ -201,17 +214,17 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
 				}
 			}
 		}
-	case constant.AllLoginButSameClassKick:
+	case constant.PcMobileAndWeb:
 		var (
-			reserved = make(map[string]struct{})
+			reserved = make(map[string]bool)
 		)
 
 		for plt, ts := range loginTokenMap {
 			if constant.PlatformIDToClass(plt) == constant.PlatformIDToClass(platformID) {
 				kickToken = append(kickToken, ts...)
 			} else {
-				if _, ok := reserved[constant.PlatformIDToClass(plt)]; !ok {
+				if !reserved[constant.PlatformIDToClass(plt)] {
-					reserved[constant.PlatformIDToClass(plt)] = struct{}{}
+					reserved[constant.PlatformIDToClass(plt)] = true
 					kickToken = append(kickToken, ts[:len(ts)-1]...)
 					continue
 				} else {
@@ -219,20 +232,36 @@ 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
+	var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
-	//if a.multiLogin.Policy == constant.Customize {
+	if a.multiLogin.Policy == constant.Customize {
-	//	adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID]
+		adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID]
-	//}
+	}
-	//l := len(adminToken)
+	l := len(adminToken)
-	//if platformID == constant.AdminPlatformID {
+	if platformID == constant.AdminPlatformID {
-	//	l++
+		l++
-	//}
+	}
-	//if l > adminTokenMaxNum {
+	if l > adminTokenMaxNum {
-	//	kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
+		kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
-	//}
+	}
 	return deleteToken, kickToken, nil
 }

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 && !errors.Is(err, redis.Nil) {
+	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return 0, 0, nil, err
 	}
 	minSeq, err := db.seqConversation.GetMinSeq(ctx, conversationID)
@@ -443,11 +443,6 @@ 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
@@ -490,7 +485,7 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
 	}
 	successMsgs, failedSeqs, err := db.msg.GetMessagesBySeq(ctx, conversationID, newSeqs)
 	if err != nil {
-		if errors.Is(err, redis.Nil) {
+		if err != redis.Nil {
 			log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
 		}
 	}
@@ -505,11 +500,6 @@ 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
 }

pkg/common/storage/controller/stream_msg.go

@@ -1,34 +0,0 @@
-package controller
-
-import (
-	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
-	"time"
-)
-
-type StreamMsgDatabase interface {
-	CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error
-	AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error
-	GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error)
-}
-
-func NewStreamMsgDatabase(db database.StreamMsg) StreamMsgDatabase {
-	return &streamMsgDatabase{db: db}
-}
-
-type streamMsgDatabase struct {
-	db database.StreamMsg
-}
-
-func (m *streamMsgDatabase) CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error {
-	return m.db.CreateStreamMsg(ctx, model)
-}
-
-func (m *streamMsgDatabase) AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error {
-	return m.db.AppendStreamMsg(ctx, clientMsgID, startIndex, packets, end, deadlineTime)
-}
-
-func (m *streamMsgDatabase) GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
-	return m.db.GetStreamMsg(ctx, clientMsgID)
-}

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: "create_time", Value: -1}})
+	opts := options.Find().SetSort(bson.D{{Key: "created_at", Value: -1}})
 
 	// Perform the search with pagination and sorting
 	return mongoutil.FindPage[*model.Group](ctx, g.coll, bson.M{

pkg/common/storage/database/mgo/stream_msg.go

@@ -1,60 +0,0 @@
-package mgo
-
-import (
-	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
-	"github.com/openimsdk/tools/db/mongoutil"
-	"github.com/openimsdk/tools/errs"
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
-	"time"
-)
-
-func NewStreamMsgMongo(db *mongo.Database) (*StreamMsgMongo, error) {
-	coll := db.Collection(database.StreamMsgName)
-	_, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
-		Keys: bson.D{
-			{Key: "client_msg_id", Value: 1},
-		},
-		Options: options.Index().SetUnique(true),
-	})
-	if err != nil {
-		return nil, errs.Wrap(err)
-	}
-	return &StreamMsgMongo{coll: coll}, nil
-}
-
-type StreamMsgMongo struct {
-	coll *mongo.Collection
-}
-
-func (m *StreamMsgMongo) CreateStreamMsg(ctx context.Context, val *model.StreamMsg) error {
-	if val.Packets == nil {
-		val.Packets = []string{}
-	}
-	return mongoutil.InsertMany(ctx, m.coll, []*model.StreamMsg{val})
-}
-
-func (m *StreamMsgMongo) AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error {
-	update := bson.M{
-		"$set": bson.M{
-			"end":           end,
-			"deadline_time": deadlineTime,
-		},
-	}
-	if len(packets) > 0 {
-		update["$push"] = bson.M{
-			"packets": bson.M{
-				"$each":     packets,
-				"$position": startIndex,
-			},
-		}
-	}
-	return mongoutil.UpdateOne(ctx, m.coll, bson.M{"client_msg_id": clientMsgID, "end": false}, update, true)
-}
-
-func (m *StreamMsgMongo) GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) {
-	return mongoutil.FindOne[*model.StreamMsg](ctx, m.coll, bson.M{"client_msg_id": clientMsgID})
-}

pkg/common/storage/database/name.go

@@ -17,5 +17,4 @@ const (
 	UserName                = "user"
 	SeqConversationName     = "seq"
 	SeqUserName             = "seq_user"
-	StreamMsgName           = "stream_msg"
 )

pkg/common/storage/database/stream_msg.go

@@ -1,13 +0,0 @@
-package database
-
-import (
-	"context"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
-	"time"
-)
-
-type StreamMsg interface {
-	CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error
-	AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error
-	GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error)
-}

pkg/common/storage/model/application.go

@@ -1,18 +0,0 @@
-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"`
-}

pkg/common/storage/model/stream_msg.go

@@ -1,21 +0,0 @@
-package model
-
-import (
-	"time"
-)
-
-const (
-	StreamMsgStatusWait = 0
-	StreamMsgStatusDone = 1
-	StreamMsgStatusFail = 2
-)
-
-type StreamMsg struct {
-	ClientMsgID    string    `bson:"client_msg_id"`
-	ConversationID string    `bson:"conversation_id"`
-	UserID         string    `bson:"user_id"`
-	Packets        []string  `bson:"packets"`
-	End            bool      `bson:"end"`
-	CreateTime     time.Time `bson:"create_time"`
-	DeadlineTime   time.Time `bson:"deadline_time"`
-}

pkg/rpccache/subscriber.go

@@ -17,18 +17,12 @@ 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

scripts/create-topic.sh

@@ -0,0 +1,55 @@
+#!/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."

scripts/mongo-init.sh

@@ -0,0 +1,66 @@
+# 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 <<EOF
+var maxRetries = 300;
+var connected = false;
+var rootUsername = '$MONGO_INITDB_ROOT_USERNAME';
+var rootPassword = '$MONGO_INITDB_ROOT_PASSWORD';
+var dbName = '$MONGO_INITDB_DATABASE';
+var openimUsername = '$MONGO_OPENIM_USERNAME';
+var openimPassword = '$MONGO_OPENIM_PASSWORD';
+
+while (!connected && maxRetries > 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
+
+
+
+

tools/changelog/changelog.go

@@ -1,198 +0,0 @@
-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)
-}

tools/changelog/main.go

@@ -0,0 +1,308 @@
+// 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
+}

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.CheckHealth(ctx, conf.Build())
+	return kafka.Check(ctx, conf.Build(), []string{conf.ToMongoTopic, conf.ToRedisTopic, conf.ToPushTopic, conf.ToOfflinePushTopic})
 }
 
 func initConfig(configDir string) (*config.Mongo, *config.Redis, *config.Kafka, *config.Minio, *config.Discovery, error) {

tools/streammsg/main.go

@@ -1,161 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"github.com/gin-gonic/gin"
-	"github.com/google/uuid"
-	"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
-	cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
-	"github.com/openimsdk/protocol/auth"
-	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/protocol/msg"
-	"github.com/openimsdk/tools/apiresp"
-	"github.com/openimsdk/tools/errs"
-	"io"
-	"net/http"
-	"strings"
-	"time"
-)
-
-const (
-	getAdminToken   = "/auth/get_admin_token"
-	sendMsgApi      = "/msg/send_msg"
-	appendStreamMsg = "/msg/append_stream_msg"
-)
-
-var (
-	ApiAddr = "http://127.0.0.1:10002"
-	Token   string
-)
-
-func ApiCall[R any](api string, req any) (*R, error) {
-	data, err := json.Marshal(req)
-	if err != nil {
-		return nil, err
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
-	defer cancel()
-	request, err := http.NewRequestWithContext(ctx, http.MethodPost, ApiAddr+api, bytes.NewBuffer(data))
-	if err != nil {
-		return nil, err
-	}
-	if Token != "" {
-		request.Header.Set("token", Token)
-	}
-	request.Header.Set(constant.OperationID, uuid.New().String())
-	response, err := http.DefaultClient.Do(request)
-	if err != nil {
-		return nil, err
-	}
-	defer response.Body.Close()
-	var resp R
-	apiResponse := apiresp.ApiResponse{
-		Data: &resp,
-	}
-	if err := json.NewDecoder(response.Body).Decode(&apiResponse); err != nil {
-		return nil, err
-	}
-	if apiResponse.ErrCode != 0 {
-		return nil, errs.NewCodeError(apiResponse.ErrCode, apiResponse.ErrMsg)
-	}
-	return &resp, nil
-}
-
-func main() {
-	resp, err := ApiCall[auth.GetAdminTokenResp](getAdminToken, &auth.GetAdminTokenReq{
-		Secret: "openIM123",
-		UserID: "imAdmin",
-	})
-	if err != nil {
-		fmt.Println("get admin token failed", err)
-		return
-	}
-	Token = resp.Token
-	g := gin.Default()
-	g.POST("/callbackExample/callbackAfterSendSingleMsgCommand", toGin(handlerUserMsg))
-	if err := g.Run(":10006"); err != nil {
-		panic(err)
-	}
-}
-
-func toGin[R any](fn func(c *gin.Context, req *R) error) gin.HandlerFunc {
-	return func(c *gin.Context) {
-		body, err := io.ReadAll(c.Request.Body)
-		if err != nil {
-			c.String(http.StatusInternalServerError, err.Error())
-			return
-		}
-		fmt.Printf("HTTP %s %s %s\n", c.Request.Method, c.Request.URL, body)
-		var req R
-		if err := json.Unmarshal(body, &req); err != nil {
-			c.String(http.StatusInternalServerError, err.Error())
-			return
-		}
-		if err := fn(c, &req); err != nil {
-			c.String(http.StatusInternalServerError, err.Error())
-			return
-		}
-		c.String(http.StatusOK, "{}")
-	}
-}
-
-func handlerUserMsg(c *gin.Context, req *cbapi.CallbackAfterSendSingleMsgReq) error {
-	if req.ContentType != constant.Text {
-		return nil
-	}
-	if !strings.Contains(req.Content, "stream") {
-		return nil
-	}
-	apiReq := apistruct.SendMsgReq{
-		RecvID: req.SendID,
-		SendMsg: apistruct.SendMsg{
-			SendID:           req.RecvID,
-			SenderNickname:   "xxx",
-			SenderFaceURL:    "",
-			SenderPlatformID: constant.AdminPlatformID,
-			ContentType:      constant.Stream,
-			SessionType:      req.SessionType,
-			SendTime:         time.Now().UnixMilli(),
-			Content: map[string]any{
-				"type":    "xxx",
-				"content": "server test stream msg",
-			},
-		},
-	}
-	go func() {
-		if err := doPushStreamMsg(&apiReq); err != nil {
-			fmt.Println("doPushStreamMsg failed", err)
-			return
-		}
-		fmt.Println("doPushStreamMsg success")
-	}()
-	return nil
-}
-
-func doPushStreamMsg(sendReq *apistruct.SendMsgReq) error {
-	resp, err := ApiCall[msg.SendMsgResp](sendMsgApi, sendReq)
-	if err != nil {
-		return err
-	}
-	const num = 5
-	for i := 1; i <= num; i++ {
-		_, err := ApiCall[msg.AppendStreamMsgResp](appendStreamMsg, &msg.AppendStreamMsgReq{
-			ClientMsgID: resp.ClientMsgID,
-			StartIndex:  int64(i - 1),
-			Packets: []string{
-				fmt.Sprintf("stream_msg_packet_%03d", i),
-			},
-			End: i == num,
-		})
-		if err != nil {
-			fmt.Println("append stream msg failed", "clientMsgID", resp.ClientMsgID, "index", fmt.Sprintf("%d/%d", i, num), "error", err)
-			return err
-		}
-		fmt.Println("append stream msg success", "clientMsgID", resp.ClientMsgID, "index", fmt.Sprintf("%d/%d", i, num))
-		time.Sleep(time.Second * 10)
-	}
-	return nil
-}

version/version

@@ -1 +1 @@
-main
+3.8.1