msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
3
0
Code Issues Projects Releases Wiki Activity

Merge branch 'openimsdk:3.8.3-patch' into 3.8.3-patch

Browse Source
pull/3763/head
chao 5 days ago committed by GitHub
parent b707cd3523 b67c03b33e
commit 3b13f3b701
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
41 changed files with 374 additions and 332 deletions
Show Stats Download Patch File Download Diff File

10
.github/workflows/auto-assign-issue.yml vendored
Unescape View File

@ -5,23 +5,23 @@ on:
jobs:
assign-issue:
if: |
contains(github.event.comment.body, '/assign') || contains(github.event.comment.body, '/accept') &&
(contains(github.event.comment.body, '/assign') || contains(github.event.comment.body, '/accept')) &&
!contains(github.event.comment.user.login, 'openim-robot')
runs-on: ubuntu-latest
permissions:
issues: write
steps:
- name: Checkout code
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Assign the issue
run: |
export LETASE_MILESTONES=$(curl 'https://api.github.com/repos/$OWNER/$PEPO/milestones' | jq -r 'last(.[]).title')
export LATEST_MILESTONE=$(curl "https://api.github.com/repos/$OWNER/$REPO/milestones" | jq -r 'last(.[]).title')
gh issue edit ${{ github.event.issue.number }} --add-assignee "${{ github.event.comment.user.login }}"
gh issue edit ${{ github.event.issue.number }} --add-label "accepted"
gh issue comment $ISSUE --body "@${{ github.event.comment.user.login }} Glad to see you accepted this issue🤲, this issue has been assigned to you. I set the milestones for this issue to [$LETASE_MILESTONES](https://github.com/$OWNER/$PEPO/milestones), We are looking forward to your PR!"
gh issue comment $ISSUE --body "@${{ github.event.comment.user.login }} Glad to see you accepted this issue🤲, this issue has been assigned to you. I set the milestones for this issue to [$LATEST_MILESTONE](https://github.com/$OWNER/$REPO/milestones), We are looking forward to your PR!"
# gh issue edit ${{ github.event.issue.number }} --milestone "$LETASE_MILESTONES"
# gh issue edit ${{ github.event.issue.number }} --milestone "$LATEST_MILESTONE"
env:
GH_TOKEN: ${{ secrets.BOT_TOKEN }}
ISSUE: ${{ github.event.issue.html_url }}

2
.github/workflows/auto-invite-comment.yml vendored
Unescape View File

@ -13,7 +13,7 @@ jobs:
steps:
- name: Invite user to join OpenIM Community
uses: peter-evans/create-or-update-comment@v4
uses: peter-evans/create-or-update-comment@v5.0.0
with:
token: ${{ secrets.BOT_GITHUB_TOKEN }}
issue-number: ${{ github.event.issue.number }}

9
.github/workflows/changelog.yml vendored
Unescape View File

@ -13,7 +13,12 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Set up Go
uses: actions/setup-go@v6.5.0
with:
go-version-file: go.mod
- name: Run Go Changelog Generator
run: |
@ -66,7 +71,7 @@ jobs:
rm -f "${{ github.event.release.tag_name }}-changelog.md"
- name: Create Pull Request
uses: peter-evans/create-pull-request@v7.0.5
uses: peter-evans/create-pull-request@v8.1.1
with:
token: ${{ secrets.GITHUB_TOKEN }}
commit-message: "Update CHANGELOG for release ${{ github.event.release.tag_name }}"

2
.github/workflows/cla-assistant.yml vendored
Unescape View File

@ -18,7 +18,7 @@ jobs:
steps:
- name: "CLA Assistant"
if: (github.event.comment.body == 'recheck' || github.event.comment.body == 'I have read the CLA Document and I hereby sign the CLA') || github.event_name == 'pull_request_target'
uses: contributor-assistant/github-action@v2.4.0
uses: contributor-assistant/github-action@v2.6.1
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
PERSONAL_ACCESS_TOKEN: ${{ secrets.BOT_TOKEN }}

12
.github/workflows/cleanup-after-milestone-prs-merged.yml vendored
Unescape View File

@ -11,14 +11,13 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4.2.0
uses: actions/checkout@v7.0.0
- name: Get the PR title and extract PR numbers
id: extract_pr_numbers
env:
PR_TITLE: ${{ github.event.pull_request.title }}
run: |
# Get the PR title
PR_TITLE="${{ github.event.pull_request.title }}"
echo "PR Title: $PR_TITLE"
# Extract PR numbers from the title
@ -57,9 +56,8 @@ jobs:
- name: Delete branch after PR close
if: steps.extract_pr_numbers.outputs.proceed == 'true' || contains(github.event.pull_request.labels.*.name, 'milestone-merge')
env:
BRANCH_NAME: ${{ github.event.pull_request.head.ref }}
run: |
BRANCH_NAME="${{ github.event.pull_request.head.ref }}"
echo "Branch to delete: $BRANCH_NAME"
git push origin --delete "$BRANCH_NAME"
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

8
.github/workflows/codeql-analysis.yml vendored
Unescape View File

@ -35,11 +35,11 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
# Initializes the CodeQL tools for scanning.
- name: Initialize CodeQL
uses: github/codeql-action/init@v3
uses: github/codeql-action/init@v4.36.2
with:
languages: ${{ matrix.language }}
# If you wish to specify custom queries, you can do so here or in a config file.
@ -50,7 +50,7 @@ jobs:
# Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
# If this step fails, then you should remove it and run the build manually (see below)
- name: Autobuild
uses: github/codeql-action/autobuild@v3
uses: github/codeql-action/autobuild@v4.36.2
# ℹ️ Command-line programs to run using the OS shell.
# 📚 https://git.io/JvXDl
@ -64,4 +64,4 @@ jobs:
# make release
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@v3
uses: github/codeql-action/analyze@v4.36.2

2
.github/workflows/comment-check.yml vendored
Unescape View File

@ -17,7 +17,7 @@ jobs:
EXCLUDE_FILES: "*.md *.txt *.html *.css *.min.js *.mdx"
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v7.0.0
- name: Search for Non-English comments
run: |

12
.github/workflows/docker-build-and-release-services-images.yml vendored
Unescape View File

@ -22,29 +22,29 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Set up QEMU
uses: docker/setup-qemu-action@v3.3.0
uses: docker/setup-qemu-action@v4.1.0
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3.8.0
uses: docker/setup-buildx-action@v4.1.0
- name: Log in to Docker Hub
uses: docker/login-action@v3.3.0
uses: docker/login-action@v4.2.0
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Log in to GitHub Container Registry
uses: docker/login-action@v3.3.0
uses: docker/login-action@v4.2.0
with:
registry: ghcr.io
username: ${{ github.repository_owner }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Log in to Aliyun Container Registry
uses: docker/login-action@v3.3.0
uses: docker/login-action@v4.2.0
with:
registry: registry.cn-hangzhou.aliyuncs.com
username: ${{ secrets.ALIREGISTRY_USERNAME }}

24
.github/workflows/go-build-test.yml vendored
Unescape View File

@ -22,14 +22,14 @@ jobs:
strategy:
matrix:
os: [ubuntu-latest]
go_version: ["1.25.x"]
go_version: ["stable"]
steps:
- name: Checkout Server repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Set up Go ${{ matrix.go_version }}
uses: actions/setup-go@v5
uses: actions/setup-go@v6.5.0
with:
go-version: ${{ matrix.go_version }}
@ -40,7 +40,7 @@ jobs:
go mod download
- name: Set up infra services
uses: hoverkraft-tech/compose-action@v2.0.1
uses: hoverkraft-tech/compose-action@v3.0.0
with:
compose-file: "./docker-compose.yml"
@ -67,7 +67,7 @@ jobs:
mage check
- name: Checkout Chat repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
with:
repository: "openimsdk/chat"
path: "chat-repo"
@ -188,21 +188,21 @@ jobs:
strategy:
matrix:
os: [ubuntu-latest]
go_version: ["1.25.x"]
go_version: ["stable"]
steps:
- name: Checkout Server repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Checkout SDK repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
with:
repository: "openimsdk/openim-sdk-core"
ref: "main"
path: ${{ env.SDK_DIR }}
- name: Set up Go ${{ matrix.go_version }}
uses: actions/setup-go@v5
uses: actions/setup-go@v6.5.0
with:
go-version: ${{ matrix.go_version }}
@ -237,14 +237,14 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
go_version: ["1.25.x"]
go_version: ["stable"]
steps:
- name: Checkout Repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Set up Go ${{ matrix.go_version }}
uses: actions/setup-go@v5
uses: actions/setup-go@v6.5.0
with:
go-version: ${{ matrix.go_version }}

2
.github/workflows/help-comment-issue.yml vendored
Unescape View File

@ -26,7 +26,7 @@ jobs:
issues: write
steps:
- name: Add comment
uses: peter-evans/create-or-update-comment@v4
uses: peter-evans/create-or-update-comment@v5.0.0
with:
issue-number: ${{ github.event.issue.number }}
token: ${{ secrets.BOT_TOKEN }}

2
.github/workflows/merge-from-milestone.yml vendored
Unescape View File

@ -40,7 +40,7 @@ jobs:
touch ${{ env.TEMP_DIR }}/created_pr_number.txt
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
with:
fetch-depth: 0
token: ${{ secrets.BOT_TOKEN }}

17
.github/workflows/publish-docker-image.yml vendored
Unescape View File

@ -14,7 +14,6 @@ on:
default: "v3.8.3"
env:
GO_VERSION: "1.22"
IMAGE_NAME: "openim-server"
# IMAGE_NAME: ${{ github.event.repository.name }}
DOCKER_BUILDKIT: 1
@ -25,22 +24,22 @@ jobs:
if: ${{ !(github.event_name == 'pull_request' && github.event.pull_request.merged == false) }}
steps:
- name: Checkout main repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
with:
path: main-repo
- name: Set up QEMU
uses: docker/setup-qemu-action@v3.3.0
uses: docker/setup-qemu-action@v4.1.0
- name: Set up Docker Buildx
id: buildx
uses: docker/setup-buildx-action@v3
uses: docker/setup-buildx-action@v4.1.0
with:
driver-opts: network=host
- name: Extract metadata for Docker
id: meta
uses: docker/metadata-action@v5.6.0
uses: docker/metadata-action@v6.1.0
with:
images: |
${{ secrets.DOCKER_USERNAME }}/${{ env.IMAGE_NAME }}
@ -80,7 +79,7 @@ jobs:
docker image ls | grep openim
- name: Checkout compose repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
with:
repository: "openimsdk/openim-docker"
path: "compose-repo"
@ -134,20 +133,20 @@ jobs:
# fi
- name: Log in to Docker Hub
uses: docker/login-action@v3.3.0
uses: docker/login-action@v4.2.0
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Log in to GitHub Container Registry
uses: docker/login-action@v3.3.0
uses: docker/login-action@v4.2.0
with:
registry: ghcr.io
username: ${{ github.repository_owner }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Log in to Aliyun Container Registry
uses: docker/login-action@v3.3.0
uses: docker/login-action@v4.2.0
with:
registry: registry.cn-hangzhou.aliyuncs.com
username: ${{ secrets.ALIREGISTRY_USERNAME }}

8
.github/workflows/remove-unused-labels.yml vendored
Unescape View File

@ -11,11 +11,11 @@ jobs:
contents: read
steps:
- name: Checkout Repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Fetch All Issues and PRs
id: fetch_issues_prs
uses: actions/github-script@v7.0.1
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
@ -38,7 +38,7 @@ jobs:
- name: Fetch All Labels
id: fetch_labels
uses: actions/github-script@v7.0.1
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
@ -52,7 +52,7 @@ jobs:
result-encoding: string
- name: Remove Unused Labels
uses: actions/github-script@v7.0.1
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |

8
.github/workflows/reopen-issue.yml vendored
Unescape View File

@ -12,11 +12,11 @@ jobs:
steps:
- name: Checkout Repository
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
- name: Fetch Closed Issues with lifecycle/stale Label
id: fetch_issues
uses: actions/github-script@v7
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
@ -40,7 +40,7 @@ jobs:
echo "Issue numbers: ${{ steps.fetch_issues.outputs.result }}"
- name: Reopen Issues
uses: actions/github-script@v7
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |
@ -59,7 +59,7 @@ jobs:
}
- name: Remove lifecycle/stale Label
uses: actions/github-script@v7
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |

4
.github/workflows/update-version-file-on-release.yml vendored
Unescape View File

@ -12,7 +12,7 @@ jobs:
steps:
# Step 1: Checkout the original repository's code
- name: Checkout code
uses: actions/checkout@v4
uses: actions/checkout@v7.0.0
with:
fetch-depth: 0
# submodules: "recursive"
@ -73,7 +73,7 @@ jobs:
# Step 7: Find and Publish Draft Release
- name: Find and Publish Draft Release
uses: actions/github-script@v7
uses: actions/github-script@v9.0.0
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
script: |

10
.github/workflows/user-first-interaction.yml vendored
Unescape View File

@ -11,11 +11,11 @@ jobs:
check_for_first_interaction:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/first-interaction@v1.3.0
- uses: actions/checkout@v7.0.0
- uses: actions/first-interaction@v3.1.0
with:
repo-token: ${{ secrets.BOT_TOKEN }}
pr-message: |
repo_token: ${{ secrets.BOT_TOKEN }}
pr_message: |
Hello! Thank you for your contribution.
If you are fixing a bug, please reference the issue number in the description.
@ -26,7 +26,7 @@ jobs:
Please leave your information in the [✨ discussions](https://github.com/orgs/OpenIMSDK/discussions/426), we expect anyone to join OpenIM developer community.
issue-message: |
issue_message: |
Hello! Thank you for filing an issue.
If this is a bug report, please include relevant logs to help us debug the problem.

2
Dockerfile
Unescape View File

@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:alpine AS builder
ARG RELEASE=false
ARG COMPRESS=false

16
build/build.sh
Unescape View File

@ -24,19 +24,13 @@ fi
cd "$BASE_DIR/.." || exit 1
split_values() {
printf '%s\n' "$1" | tr ', ' '\n\n' | while IFS= read -r value; do
[[ -n "$value" ]] && printf '%s\n' "$value"
done
}
print_command() {
printf '%q ' "$@"
printf '\n'
echo "$1" | grep -o '[^, ]\+'
}
run_or_print() {
if [[ "$DRY_RUN" == "true" ]]; then
print_command "$@"
printf '%q ' "$@"
printf '\n'
else
"$@"
fi
@ -44,7 +38,9 @@ run_or_print() {
build_local() {
echo -e "${CYAN}Building all services...${NO_COLOR}"
RELEASE="$RELEASE" docker compose -f "$COMPOSE_FILE" build
while IFS= read -r service; do
RELEASE="$RELEASE" docker compose -f "$COMPOSE_FILE" build "$service"
done < <(docker compose -f "$COMPOSE_FILE" config --services)
echo -e "${CYAN}Tagging compatibility images for Kubernetes...${NO_COLOR}"
while IFS= read -r built_image; do

10
build/images/openim-server/Dockerfile
Unescape View File

@ -1,8 +1,4 @@
FROM golang:1.25-alpine AS builder
ARG CMD_PATH
ARG BINARY_NAME
ARG RELEASE=false
FROM golang:alpine AS builder
ENV SERVER_DIR=/openim-server
WORKDIR $SERVER_DIR
@ -10,6 +6,10 @@ WORKDIR $SERVER_DIR
COPY . .
RUN go mod tidy
ARG CMD_PATH
ARG BINARY_NAME
ARG RELEASE=false
RUN if [ "$RELEASE" = "true" ]; then \
go build -trimpath -ldflags "-s -w" -o _output/${BINARY_NAME} ./${CMD_PATH}; \
else \

72
build/images/openim-server/docker-compose.build.override.yml
Unescape View File

@ -1,72 +0,0 @@
services:
openim-api:
build:
platforms:
- linux/amd64
- linux/arm64
openim-crontask:
build:
platforms:
- linux/amd64
- linux/arm64
openim-msggateway:
build:
platforms:
- linux/amd64
- linux/arm64
openim-msgtransfer:
build:
platforms:
- linux/amd64
- linux/arm64
openim-push:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-auth:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-conversation:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-friend:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-group:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-msg:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-third:
build:
platforms:
- linux/amd64
- linux/arm64
openim-rpc-user:
build:
platforms:
- linux/amd64
- linux/arm64

2
go.mod
Unescape View File

@ -13,7 +13,7 @@ require (
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/protocol v0.0.73-alpha.12
github.com/openimsdk/tools v0.0.50-alpha.117
github.com/openimsdk/tools v0.0.50-alpha.119
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0
github.com/stretchr/testify v1.11.1

2
go.sum
Unescape View File

@ -365,6 +365,8 @@ github.com/openimsdk/protocol v0.0.73-alpha.12 h1:2NYawXeHChYUeSme6QJ9pOLh+Empce
github.com/openimsdk/protocol v0.0.73-alpha.12/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/tools v0.0.50-alpha.117 h1:ACfijEVCeBcttT7OOkNGOOOvq14pJtb9szNIMHLm6Vc=
github.com/openimsdk/tools v0.0.50-alpha.117/go.mod h1:I0WESSa7ghPIo9BL+ETlH/qEIbO6+KZioM1jwNuDwz0=
github.com/openimsdk/tools v0.0.50-alpha.119 h1:S/RjRtL0ciwiG6pZKssbU//qVSWi7AuKHOPCHsQgz68=
github.com/openimsdk/tools v0.0.50-alpha.119/go.mod h1:I0WESSa7ghPIo9BL+ETlH/qEIbO6+KZioM1jwNuDwz0=
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=

18
internal/msggateway/ws_server.go
Unescape View File

@ -14,6 +14,7 @@ import (
"github.com/openimsdk/tools/apiresp"
"github.com/go-playground/validator/v10"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
@ -288,25 +289,12 @@ func (ws *WsServer) registerClient(client *Client) {
}
}
wg := sync.WaitGroup{}
log.ZDebug(client.ctx, "ws.msgGatewayConfig.Discovery.Enable", "discoveryEnable", ws.msgGatewayConfig.Discovery.Enable)
if ws.msgGatewayConfig.Discovery.Enable != "k8s" {
wg.Add(1)
go func() {
defer wg.Done()
_ = ws.sendUserOnlineInfoToOtherNode(client.ctx, client)
}()
if ws.msgGatewayConfig.Discovery.Enable != config.KUBERNETES {
_ = ws.sendUserOnlineInfoToOtherNode(client.ctx, client)
}
//wg.Add(1)
//go func() {
// defer wg.Done()
// ws.SetUserOnlineStatus(client.ctx, client, constant.Online)
//}()
wg.Wait()
log.ZDebug(client.ctx, "user online", "online user Num", ws.onlineUserNum.Load(), "online user conn Num", ws.onlineUserConnNum.Load())
}

7
internal/push/onlinepusher.go
Unescape View File

@ -4,6 +4,7 @@ import (
"context"
"sync"
conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/protocol/msggateway"
"github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/discovery"
@ -39,11 +40,11 @@ func (u emptyOnlinePusher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *
func NewOnlinePusher(disCov discovery.SvcDiscoveryRegistry, config *Config) OnlinePusher {
switch config.Discovery.Enable {
case "k8s":
return NewK8sStaticConsistentHash(disCov, config)
case conf.KUBERNETES:
return NewDefaultAllNode(disCov, config)
case "zookeeper":
return NewDefaultAllNode(disCov, config)
case "etcd":
case conf.ETCD:
return NewDefaultAllNode(disCov, config)
default:
return newEmptyOnlinePusher()

99
internal/rpc/conversation/conversation.go
Unescape View File

@ -24,7 +24,6 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
dbModel "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
@ -233,9 +232,28 @@ func (c *conversationServer) getConversations(ctx context.Context, ownerUserID s
// Deprecated
func (c *conversationServer) SetConversation(ctx context.Context, req *pbconversation.SetConversationReq) (*pbconversation.SetConversationResp, error) {
if req.Conversation == nil {
return nil, errs.ErrArgs.WrapMsg("conversation must not be nil")
}
var conversation dbModel.Conversation
if err := datautil.CopyStructFields(&conversation, req.Conversation); err != nil {
return nil, err
conversation.OwnerUserID = req.Conversation.OwnerUserID
conversation.ConversationID = req.Conversation.ConversationID
conversation.RecvMsgOpt = req.Conversation.RecvMsgOpt
conversation.ConversationType = req.Conversation.ConversationType
conversation.UserID = req.Conversation.UserID
conversation.GroupID = req.Conversation.GroupID
conversation.IsPinned = req.Conversation.IsPinned
conversation.AttachedInfo = req.Conversation.AttachedInfo
conversation.IsPrivateChat = req.Conversation.IsPrivateChat
conversation.GroupAtType = req.Conversation.GroupAtType
conversation.Ex = req.Conversation.Ex
conversation.BurnDuration = req.Conversation.BurnDuration
conversation.MinSeq = req.Conversation.MinSeq
conversation.MaxSeq = req.Conversation.MaxSeq
conversation.MsgDestructTime = req.Conversation.MsgDestructTime
conversation.IsMsgDestruct = req.Conversation.IsMsgDestruct
if req.Conversation.LatestMsgDestructTime != 0 {
conversation.LatestMsgDestructTime = time.UnixMilli(req.Conversation.LatestMsgDestructTime)
}
err := c.conversationDatabase.SetUserConversations(ctx, req.Conversation.OwnerUserID, []*dbModel.Conversation{&conversation})
if err != nil {
@ -506,17 +524,6 @@ func (c *conversationServer) GetUserConversationIDsHash(ctx context.Context, req
return &pbconversation.GetUserConversationIDsHashResp{Hash: hash}, nil
}
func (c *conversationServer) GetConversationsByConversationID(
ctx context.Context,
req *pbconversation.GetConversationsByConversationIDReq,
) (*pbconversation.GetConversationsByConversationIDResp, error) {
conversations, err := c.conversationDatabase.GetConversationsByConversationID(ctx, req.ConversationIDs)
if err != nil {
return nil, err
}
return &pbconversation.GetConversationsByConversationIDResp{Conversations: convert.ConversationsDB2Pb(conversations)}, nil
}
func (c *conversationServer) GetConversationOfflinePushUserIDs(ctx context.Context, req *pbconversation.GetConversationOfflinePushUserIDsReq) (*pbconversation.GetConversationOfflinePushUserIDsResp, error) {
if req.ConversationID == "" {
return nil, errs.ErrArgs.WrapMsg("conversationID is empty")
@ -606,9 +613,17 @@ func (c *conversationServer) getConversationInfo(
}
for conversationID, chatLog := range chatLogs {
pbchatLog := &pbconversation.ConversationElem{}
msgInfo := &pbconversation.MsgInfo{}
if err := datautil.CopyStructFields(msgInfo, chatLog); err != nil {
return nil, err
msgInfo := &pbconversation.MsgInfo{
ServerMsgID: chatLog.ServerMsgID,
ClientMsgID: chatLog.ClientMsgID,
SessionType: chatLog.SessionType,
SendID: chatLog.SendID,
RecvID: chatLog.RecvID,
GroupID: chatLog.GroupID,
MsgFrom: chatLog.MsgFrom,
ContentType: chatLog.ContentType,
Content: string(chatLog.Content),
Ex: chatLog.Ex,
}
switch chatLog.SessionType {
case constant.SingleChatType:
@ -708,56 +723,6 @@ func (c *conversationServer) GetOwnerConversation(ctx context.Context, req *pbco
}, nil
}
func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _ *pbconversation.GetConversationsNeedClearMsgReq) (*pbconversation.GetConversationsNeedClearMsgResp, error) {
num, err := c.conversationDatabase.GetAllConversationIDsNumber(ctx)
if err != nil {
log.ZError(ctx, "GetAllConversationIDsNumber failed", err)
return nil, err
}
const batchNum = 100
if num == 0 {
return nil, errs.New("Need Destruct Msg is nil").Wrap()
}
maxPage := (num + batchNum - 1) / batchNum
temp := make([]*model.Conversation, 0, maxPage*batchNum)
for pageNumber := 0; pageNumber < int(maxPage); pageNumber++ {
pagination := &sdkws.RequestPagination{
PageNumber: int32(pageNumber),
ShowNumber: batchNum,
}
conversationIDs, err := c.conversationDatabase.PageConversationIDs(ctx, pagination)
if err != nil {
log.ZError(ctx, "PageConversationIDs failed", err, "pageNumber", pageNumber)
continue
}
// log.ZDebug(ctx, "PageConversationIDs success", "pageNumber", pageNumber, "conversationIDsNum", len(conversationIDs), "conversationIDs", conversationIDs)
if len(conversationIDs) == 0 {
continue
}
conversations, err := c.conversationDatabase.GetConversationsByConversationID(ctx, conversationIDs)
if err != nil {
log.ZError(ctx, "GetConversationsByConversationID failed", err, "conversationIDs", conversationIDs)
continue
}
for _, conversation := range conversations {
if conversation.IsMsgDestruct && conversation.MsgDestructTime != 0 && ((time.Now().UnixMilli() > (conversation.MsgDestructTime + conversation.LatestMsgDestructTime.UnixMilli() + 8*60*60)) || // 8*60*60 is UTC+8
conversation.LatestMsgDestructTime.IsZero()) {
temp = append(temp, conversation)
}
}
}
return &pbconversation.GetConversationsNeedClearMsgResp{Conversations: convert.ConversationsDB2Pb(temp)}, nil
}
func (c *conversationServer) GetNotNotifyConversationIDs(ctx context.Context, req *pbconversation.GetNotNotifyConversationIDsReq) (*pbconversation.GetNotNotifyConversationIDsResp, error) {
conversationIDs, err := c.conversationDatabase.GetNotNotifyConversationIDs(ctx, req.UserID)
if err != nil {

9
internal/rpc/msg/delete.go
Unescape View File

@ -19,7 +19,6 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/log"
@ -74,7 +73,7 @@ func (m *msgServer) DeleteMsgs(ctx context.Context, req *msg.DeleteMsgsReq) (*ms
if err := m.MsgDatabase.DeleteMsgsPhysicalBySeqs(ctx, req.ConversationID, req.Seqs); err != nil {
return nil, err
}
conv, err := m.conversationClient.GetConversationsByConversationID(ctx, req.ConversationID)
conv, err := m.conversationClient.GetConversation(ctx, req.ConversationID, req.UserID)
if err != nil {
return nil, err
}
@ -113,14 +112,12 @@ func (m *msgServer) DeleteMsgPhysical(ctx context.Context, req *msg.DeleteMsgPhy
}
func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []string, userID string, deleteSyncOpt *msg.DeleteSyncOpt) error {
conversations, err := m.conversationClient.GetConversationsByConversationIDs(ctx, conversationIDs)
conversations, err := m.conversationClient.GetConversations(ctx, conversationIDs, userID)
if err != nil {
return err
}
var existConversations []*conversation.Conversation
var existConversationIDs []string
for _, conversation := range conversations {
existConversations = append(existConversations, conversation)
existConversationIDs = append(existConversationIDs, conversation.ConversationID)
}
log.ZDebug(ctx, "ClearConversationsMsg", "existConversationIDs", existConversationIDs)
@ -149,7 +146,7 @@ func (m *msgServer) clearConversation(ctx context.Context, conversationIDs []str
if err := m.MsgDatabase.SetMinSeqs(ctx, m.getMinSeqs(maxSeqs)); err != nil {
return err
}
for _, conversation := range existConversations {
for _, conversation := range conversations {
tips := &sdkws.ClearConversationTips{UserID: userID, ConversationIDs: []string{conversation.ConversationID}}
m.notificationSender.NotificationWithSessionType(ctx, userID, m.conversationAndGetRecvID(conversation, userID), constant.ClearConversationNotification, conversation.ConversationType, tips)
}

22
internal/rpc/msg/sync_msg.go
Unescape View File

@ -24,7 +24,6 @@ import (
"github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/timeutil"
)
@ -216,9 +215,24 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
// Construct response with updated information
for _, chatLog := range chatLogs {
pbchatLog := &msg.ChatLog{}
datautil.CopyStructFields(pbchatLog, chatLog.MsgData)
pbchatLog.SendTime = chatLog.MsgData.SendTime
pbchatLog.CreateTime = chatLog.MsgData.CreateTime
msgData := chatLog.MsgData
pbchatLog.ServerMsgID = msgData.ServerMsgID
pbchatLog.ClientMsgID = msgData.ClientMsgID
pbchatLog.SendID = msgData.SendID
pbchatLog.RecvID = msgData.RecvID
pbchatLog.GroupID = msgData.GroupID
pbchatLog.SenderPlatformID = msgData.SenderPlatformID
pbchatLog.SenderNickname = msgData.SenderNickname
pbchatLog.SenderFaceURL = msgData.SenderFaceURL
pbchatLog.SessionType = msgData.SessionType
pbchatLog.MsgFrom = msgData.MsgFrom
pbchatLog.ContentType = msgData.ContentType
pbchatLog.Content = string(msgData.Content)
pbchatLog.Status = msgData.Status
pbchatLog.SendTime = msgData.SendTime
pbchatLog.CreateTime = msgData.CreateTime
pbchatLog.Ex = msgData.Ex
pbchatLog.Seq = msgData.Seq
if chatLog.MsgData.SenderNickname == "" {
pbchatLog.SenderNickname = sendMap[chatLog.MsgData.SendID]
}

2
internal/tools/cron_test.go
Unescape View File

@ -23,7 +23,7 @@ func TestName(t *testing.T) {
Address: []string{"localhost:12379"},
},
}
client, err := kdisc.NewDiscoveryRegister(conf, "source")
client, err := kdisc.NewDiscoveryRegister(conf, &config.Share{}, nil)
if err != nil {
panic(err)
}

11
pkg/common/config/config.go
Unescape View File

@ -484,9 +484,14 @@ type ZooKeeper struct {
}
type Discovery struct {
Enable string `mapstructure:"enable"`
Etcd Etcd `mapstructure:"etcd"`
ZooKeeper ZooKeeper `mapstructure:"zooKeeper"`
Enable string `mapstructure:"enable"`
Etcd Etcd `mapstructure:"etcd"`
Kubernetes Kubernetes `mapstructure:"kubernetes"`
ZooKeeper ZooKeeper `mapstructure:"zooKeeper"`
}
type Kubernetes struct {
Namespace string `mapstructure:"namespace"`
}
type Etcd struct {

24
pkg/common/config/load_config_test.go
Unescape View File

@ -1,8 +1,12 @@
package config
import (
"github.com/stretchr/testify/assert"
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
)
func TestLoadLogConfig(t *testing.T) {
@ -27,6 +31,24 @@ func TestLoadWebhooksConfig(t *testing.T) {
}
func TestLoadDiscoveryKubernetesConfig(t *testing.T) {
path := filepath.Join(t.TempDir(), "discovery.yml")
err := os.WriteFile(path, []byte(`enable: kubernetes
kubernetes:
namespace: openim
etcd:
rootDirectory: openim
address: [localhost:12379]
`), 0600)
assert.Nil(t, err)
var discovery Discovery
err = LoadConfig(path, "IMENV_DISCOVERY", &discovery)
assert.Nil(t, err)
assert.Equal(t, KUBERNETES, discovery.Enable)
assert.Equal(t, "openim", discovery.Kubernetes.Namespace)
}
func TestLoadOpenIMRpcUserConfig(t *testing.T) {
var user User
err := LoadConfig("../../../config/openim-rpc-user.yml", "IMENV_OPENIM_RPC_USER", &user)

98
pkg/common/convert/conversation.go
Unescape View File

@ -15,46 +15,120 @@
package convert
import (
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/tools/utils/datautil"
)
func ConversationDB2Pb(conversationDB *model.Conversation) *conversation.Conversation {
conversationPB := &conversation.Conversation{}
conversationPB.LatestMsgDestructTime = conversationDB.LatestMsgDestructTime.UnixMilli()
if err := datautil.CopyStructFields(conversationPB, conversationDB); err != nil {
if conversationDB == nil {
return nil
}
return conversationPB
return &conversation.Conversation{
OwnerUserID: conversationDB.OwnerUserID,
ConversationID: conversationDB.ConversationID,
RecvMsgOpt: conversationDB.RecvMsgOpt,
ConversationType: conversationDB.ConversationType,
UserID: conversationDB.UserID,
GroupID: conversationDB.GroupID,
IsPinned: conversationDB.IsPinned,
AttachedInfo: conversationDB.AttachedInfo,
IsPrivateChat: conversationDB.IsPrivateChat,
GroupAtType: conversationDB.GroupAtType,
Ex: conversationDB.Ex,
BurnDuration: conversationDB.BurnDuration,
MinSeq: conversationDB.MinSeq,
MaxSeq: conversationDB.MaxSeq,
MsgDestructTime: conversationDB.MsgDestructTime,
LatestMsgDestructTime: conversationDB.LatestMsgDestructTime.UnixMilli(),
IsMsgDestruct: conversationDB.IsMsgDestruct,
}
}
func ConversationsDB2Pb(conversationsDB []*model.Conversation) (conversationsPB []*conversation.Conversation) {
for _, conversationDB := range conversationsDB {
conversationPB := &conversation.Conversation{}
if err := datautil.CopyStructFields(conversationPB, conversationDB); err != nil {
if conversationDB == nil {
continue
}
conversationPB.LatestMsgDestructTime = conversationDB.LatestMsgDestructTime.UnixMilli()
conversationPB := &conversation.Conversation{
OwnerUserID: conversationDB.OwnerUserID,
ConversationID: conversationDB.ConversationID,
RecvMsgOpt: conversationDB.RecvMsgOpt,
ConversationType: conversationDB.ConversationType,
UserID: conversationDB.UserID,
GroupID: conversationDB.GroupID,
IsPinned: conversationDB.IsPinned,
AttachedInfo: conversationDB.AttachedInfo,
IsPrivateChat: conversationDB.IsPrivateChat,
GroupAtType: conversationDB.GroupAtType,
Ex: conversationDB.Ex,
BurnDuration: conversationDB.BurnDuration,
MinSeq: conversationDB.MinSeq,
MaxSeq: conversationDB.MaxSeq,
MsgDestructTime: conversationDB.MsgDestructTime,
LatestMsgDestructTime: conversationDB.LatestMsgDestructTime.UnixMilli(),
IsMsgDestruct: conversationDB.IsMsgDestruct,
}
conversationsPB = append(conversationsPB, conversationPB)
}
return conversationsPB
}
func ConversationPb2DB(conversationPB *conversation.Conversation) *model.Conversation {
conversationDB := &model.Conversation{}
if err := datautil.CopyStructFields(conversationDB, conversationPB); err != nil {
if conversationPB == nil {
return nil
}
conversationDB := &model.Conversation{
OwnerUserID: conversationPB.OwnerUserID,
ConversationID: conversationPB.ConversationID,
RecvMsgOpt: conversationPB.RecvMsgOpt,
ConversationType: conversationPB.ConversationType,
UserID: conversationPB.UserID,
GroupID: conversationPB.GroupID,
IsPinned: conversationPB.IsPinned,
AttachedInfo: conversationPB.AttachedInfo,
IsPrivateChat: conversationPB.IsPrivateChat,
GroupAtType: conversationPB.GroupAtType,
Ex: conversationPB.Ex,
BurnDuration: conversationPB.BurnDuration,
MinSeq: conversationPB.MinSeq,
MaxSeq: conversationPB.MaxSeq,
MsgDestructTime: conversationPB.MsgDestructTime,
IsMsgDestruct: conversationPB.IsMsgDestruct,
}
if conversationPB.LatestMsgDestructTime != 0 {
conversationDB.LatestMsgDestructTime = time.UnixMilli(conversationPB.LatestMsgDestructTime)
}
return conversationDB
}
func ConversationsPb2DB(conversationsPB []*conversation.Conversation) (conversationsDB []*model.Conversation) {
for _, conversationPB := range conversationsPB {
conversationDB := &model.Conversation{}
if err := datautil.CopyStructFields(conversationDB, conversationPB); err != nil {
if conversationPB == nil {
continue
}
conversationDB := &model.Conversation{
OwnerUserID: conversationPB.OwnerUserID,
ConversationID: conversationPB.ConversationID,
RecvMsgOpt: conversationPB.RecvMsgOpt,
ConversationType: conversationPB.ConversationType,
UserID: conversationPB.UserID,
GroupID: conversationPB.GroupID,
IsPinned: conversationPB.IsPinned,
AttachedInfo: conversationPB.AttachedInfo,
IsPrivateChat: conversationPB.IsPrivateChat,
GroupAtType: conversationPB.GroupAtType,
Ex: conversationPB.Ex,
BurnDuration: conversationPB.BurnDuration,
MinSeq: conversationPB.MinSeq,
MaxSeq: conversationPB.MaxSeq,
MsgDestructTime: conversationPB.MsgDestructTime,
IsMsgDestruct: conversationPB.IsMsgDestruct,
}
if conversationPB.LatestMsgDestructTime != 0 {
conversationDB.LatestMsgDestructTime = time.UnixMilli(conversationPB.LatestMsgDestructTime)
}
conversationsDB = append(conversationsDB, conversationDB)
}
return conversationsDB

22
pkg/common/convert/friend.go
Unescape View File

@ -21,18 +21,22 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/openimsdk/open-im-server/v3/pkg/notification/common_user"
"github.com/openimsdk/protocol/relation"
"github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/timeutil"
)
func FriendPb2DB(friend *sdkws.FriendInfo) *model.Friend {
dbFriend := &model.Friend{}
err := datautil.CopyStructFields(dbFriend, friend)
if err != nil {
if friend == nil {
return nil
}
dbFriend := &model.Friend{}
dbFriend.OwnerUserID = friend.OwnerUserID
dbFriend.Remark = friend.Remark
dbFriend.AddSource = friend.AddSource
dbFriend.OperatorUserID = friend.OperatorUserID
dbFriend.Ex = friend.Ex
dbFriend.IsPinned = friend.IsPinned
dbFriend.FriendUserID = friend.FriendUser.UserID
dbFriend.CreateTime = timeutil.UnixSecondToTime(friend.CreateTime)
return dbFriend
@ -69,10 +73,12 @@ func FriendsDB2Pb(ctx context.Context, friendsDB []*model.Friend, getUsers func(
}
for _, friend := range friendsDB {
friendPb := &sdkws.FriendInfo{FriendUser: &sdkws.UserInfo{}}
err := datautil.CopyStructFields(friendPb, friend)
if err != nil {
return nil, err
}
friendPb.OwnerUserID = friend.OwnerUserID
friendPb.Remark = friend.Remark
friendPb.AddSource = friend.AddSource
friendPb.OperatorUserID = friend.OperatorUserID
friendPb.Ex = friend.Ex
friendPb.IsPinned = friend.IsPinned
friendPb.FriendUser.UserID = users[friend.FriendUserID].UserID
friendPb.FriendUser.Nickname = users[friend.FriendUserID].Nickname

15
pkg/common/discoveryregister/discoveryregister.go
Unescape View File

@ -22,19 +22,26 @@ import (
"github.com/openimsdk/tools/discovery/etcd"
"github.com/openimsdk/tools/discovery/kubernetes"
"github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/utils/runtimeenv"
"google.golang.org/grpc"
)
// NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type.
func NewDiscoveryRegister(discovery *config.Discovery, share *config.Share, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
switch discovery.Enable {
case "k8s":
return kubernetes.NewConnManager("default", watchNames,
if runtimeenv.RuntimeEnvironment() == config.KUBERNETES {
namespace := discovery.Kubernetes.Namespace
if namespace == "" {
namespace = "default"
}
return kubernetes.NewConnManager(namespace, watchNames,
grpc.WithDefaultCallOptions(
grpc.MaxCallSendMsgSize(1024*1024*20),
),
)
case "etcd":
}
switch discovery.Enable {
case config.ETCD:
return etcd.NewSvcDiscoveryRegistry(
discovery.Etcd.RootDirectory,
discovery.Etcd.Address,

34
pkg/common/discoveryregister/discoveryregister_test.go
Unescape View File

@ -15,16 +15,12 @@
package discoveryregister
import (
"os"
)
"strings"
"testing"
func setupTestEnvironment() {
os.Setenv("ZOOKEEPER_SCHEMA", "openim")
os.Setenv("ZOOKEEPER_ADDRESS", "172.28.0.1")
os.Setenv("ZOOKEEPER_PORT", "12181")
os.Setenv("ZOOKEEPER_USERNAME", "")
os.Setenv("ZOOKEEPER_PASSWORD", "")
}
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/tools/utils/runtimeenv"
)
//func TestNewDiscoveryRegister(t *testing.T) {
// setupTestEnvironment()
@ -58,3 +54,23 @@ func setupTestEnvironment() {
// }
// }
//}
func TestNewDiscoveryRegisterRejectsKubernetesOutsideCluster(t *testing.T) {
if runtimeenv.RuntimeEnvironment() == config.KUBERNETES {
t.Skip("outside-cluster fallback is only relevant outside Kubernetes")
}
discovery := &config.Discovery{
Enable: config.KUBERNETES,
Kubernetes: config.Kubernetes{
Namespace: "default",
},
}
client, err := NewDiscoveryRegister(discovery, &config.Share{}, nil)
if err == nil && client != nil {
client.Close()
}
if err == nil || !strings.Contains(err.Error(), "unsupported discovery type") {
t.Fatalf("%q outside Kubernetes should not select Kubernetes discovery: %v", config.KUBERNETES, err)
}
}

13
pkg/common/storage/controller/conversation.go
Unescape View File

@ -59,8 +59,6 @@ type ConversationDatabase interface {
GetAllConversationIDsNumber(ctx context.Context) (int64, error)
// PageConversationIDs paginates through conversation IDs based on the specified pagination settings.
PageConversationIDs(ctx context.Context, pagination pagination.Pagination) (conversationIDs []string, err error)
// GetConversationsByConversationID retrieves conversations by their IDs.
GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.Conversation, error)
// GetConversationIDsNeedDestruct fetches conversations that need to be destructed based on specific criteria.
GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.Conversation, error)
// GetConversationNotReceiveMessageUserIDs gets user IDs for users in a conversation who have not received messages.
@ -127,13 +125,10 @@ func (c *conversationDatabase) SetUsersConversationFieldTx(ctx context.Context,
var conversations []*relationtb.Conversation
now := time.Now()
for _, v := range NotUserIDs {
temp := new(relationtb.Conversation)
if err = datautil.CopyStructFields(temp, conversation); err != nil {
return err
}
temp := *conversation
temp.OwnerUserID = v
temp.CreateTime = now
conversations = append(conversations, temp)
conversations = append(conversations, &temp)
}
if len(conversations) > 0 {
err = c.conversationDB.Create(ctx, conversations)
@ -351,10 +346,6 @@ func (c *conversationDatabase) PageConversationIDs(ctx context.Context, paginati
return c.conversationDB.PageConversationIDs(ctx, pagination)
}
func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.Conversation, error) {
return c.conversationDB.GetConversationsByConversationID(ctx, conversationIDs)
}
func (c *conversationDatabase) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.Conversation, error) {
return c.conversationDB.GetConversationIDsNeedDestruct(ctx)
}

1
pkg/common/storage/database/conversation.go
Unescape View File

@ -38,7 +38,6 @@ type Conversation interface {
GetAllConversationIDs(ctx context.Context) ([]string, error)
GetAllConversationIDsNumber(ctx context.Context) (int64, error)
PageConversationIDs(ctx context.Context, pagination pagination.Pagination) (conversationIDs []string, err error)
GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*model.Conversation, error)
GetConversationIDsNeedDestruct(ctx context.Context) ([]*model.Conversation, error)
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
FindConversationUserVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error)

27
pkg/common/storage/database/mgo/conversation.go
Unescape View File

@ -32,12 +32,25 @@ import (
func NewConversationMongo(db *mongo.Database) (*ConversationMgo, error) {
coll := db.Collection(database.ConversationName)
_, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
Keys: bson.D{
{Key: "owner_user_id", Value: 1},
{Key: "conversation_id", Value: 1},
_, err := coll.Indexes().CreateMany(context.Background(), []mongo.IndexModel{
{
Keys: bson.D{
{Key: "owner_user_id", Value: 1},
{Key: "conversation_id", Value: 1},
},
Options: options.Index().SetUnique(true),
},
{
Keys: bson.D{
{Key: "user_id", Value: 1},
},
Options: options.Index(),
},
{
Keys: bson.D{
{Key: "conversation_id", Value: 1},
},
},
Options: options.Index().SetUnique(true),
})
if err != nil {
return nil, errs.Wrap(err)
@ -191,10 +204,6 @@ func (c *ConversationMgo) PageConversationIDs(ctx context.Context, pagination pa
return mongoutil.FindPageOnly[string](ctx, c.coll, bson.M{}, pagination, options.Find().SetProjection(bson.M{"conversation_id": 1}))
}
func (c *ConversationMgo) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*model.Conversation, error) {
return mongoutil.Find[*model.Conversation](ctx, c.coll, bson.M{"conversation_id": bson.M{"$in": conversationIDs}})
}
func (c *ConversationMgo) GetConversationIDsNeedDestruct(ctx context.Context) ([]*model.Conversation, error) {
// "is_msg_destruct = 1 && msg_destruct_time != 0 && (UNIX_TIMESTAMP(NOW()) > (msg_destruct_time + UNIX_TIMESTAMP(latest_msg_destruct_time)) || latest_msg_destruct_time is NULL)"
return mongoutil.Find[*model.Conversation](ctx, c.coll, bson.M{

2
pkg/rpccache/online.go
Unescape View File

@ -202,8 +202,6 @@ func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string)
if err != nil {
return nil, err
}
tmp := make([]int32, len(platformIDs))
copy(tmp, platformIDs)
return platformIDs, nil
}

13
pkg/rpcli/conversation.go
Unescape View File

@ -2,6 +2,7 @@ package rpcli
import (
"context"
"github.com/openimsdk/protocol/conversation"
"google.golang.org/grpc"
)
@ -30,18 +31,6 @@ func (x *ConversationClient) SetConversations(ctx context.Context, ownerUserIDs
return ignoreResp(x.ConversationClient.SetConversations(ctx, req))
}
func (x *ConversationClient) GetConversationsByConversationIDs(ctx context.Context, conversationIDs []string) ([]*conversation.Conversation, error) {
if len(conversationIDs) == 0 {
return nil, nil
}
req := &conversation.GetConversationsByConversationIDReq{ConversationIDs: conversationIDs}
return extractField(ctx, x.ConversationClient.GetConversationsByConversationID, req, (*conversation.GetConversationsByConversationIDResp).GetConversations)
}
func (x *ConversationClient) GetConversationsByConversationID(ctx context.Context, conversationID string) (*conversation.Conversation, error) {
return firstValue(x.GetConversationsByConversationIDs(ctx, []string{conversationID}))
}
func (x *ConversationClient) SetConversationMinSeq(ctx context.Context, conversationID string, ownerUserIDs []string, minSeq int64) error {
if len(ownerUserIDs) == 0 {
return nil

12
pkg/tools/batcher/batcher.go
Unescape View File

@ -149,12 +149,21 @@ func (b *Batcher[T]) Put(ctx context.Context, data *T) error {
func (b *Batcher[T]) scheduler() {
ticker := time.NewTicker(b.config.interval)
// Track whether b.data was closed by an external caller so the
// cleanup below does not close it a second time. The only routes
// out of this function that leave b.data open are the nil-message
// and ticker paths; the ok == false branch means someone already
// closed the channel, and calling close(b.data) again there would
// panic with "close of closed channel" (#3653).
externallyClosed := false
defer func() {
ticker.Stop()
for _, ch := range b.chArrays {
close(ch)
}
close(b.data)
if !externallyClosed {
close(b.data)
}
b.wait.Done()
}()
@ -167,6 +176,7 @@ func (b *Batcher[T]) scheduler() {
case data, ok := <-b.data:
if !ok {
// If the data channel is closed unexpectedly
externallyClosed = true
return
}
if data == nil {

41
tools/changelog/changelog.go
Unescape View File

@ -2,6 +2,7 @@ package main
import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
@ -35,6 +36,10 @@ type ReleaseData struct {
Published string `json:"published_at"`
}
type GitHubError struct {
Message string `json:"message"`
}
// Method to classify and format release notes.
func (g *GitHubRepo) classifyReleaseNotes(body string) map[string][]string {
result := map[string][]string{
@ -49,10 +54,7 @@ func (g *GitHubRepo) classifyReleaseNotes(body string) map[string][]string {
// 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 {
for line := range strings.SplitSeq(body, "\n") {
// Skip lines that contain "deps: Merge"
if strings.Contains(strings.ToLower(line), "deps: merge #") {
continue
@ -108,7 +110,17 @@ func (g *GitHubRepo) classifyReleaseNotes(body string) map[string][]string {
}
// Method to generate the final changelog.
func (g *GitHubRepo) generateChangelog(tag, date, htmlURL, body string) string {
func (g *GitHubRepo) generateChangelog(tag, date, htmlURL, body string) (string, error) {
if tag == "" {
return "", errors.New("release tag is empty")
}
if len(date) < 10 {
return "", fmt.Errorf("release published_at is invalid: %q", date)
}
if htmlURL == "" {
return "", errors.New("release html_url is empty")
}
sections := g.classifyReleaseNotes(body)
// Convert ISO 8601 date to simpler format (YYYY-MM-DD)
@ -140,7 +152,7 @@ func (g *GitHubRepo) generateChangelog(tag, date, htmlURL, body string) string {
changelog += fmt.Sprintf("**Full Changelog**: %s\n", g.FullChangelog)
}
return changelog
return changelog, nil
}
// Method to fetch release data from GitHub API.
@ -165,6 +177,13 @@ func (g *GitHubRepo) fetchReleaseData(version string) (*ReleaseData, error) {
if err != nil {
return nil, err
}
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
var githubError GitHubError
if err := json.Unmarshal(body, &githubError); err == nil && githubError.Message != "" {
return nil, fmt.Errorf("github api returned %s: %s", resp.Status, githubError.Message)
}
return nil, fmt.Errorf("github api returned %s", resp.Status)
}
var releaseData ReleaseData
err = json.Unmarshal(body, &releaseData)
@ -188,11 +207,15 @@ func main() {
// 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
fmt.Fprintln(os.Stderr, "Error fetching release data:", err)
os.Exit(1)
}
// Generate and print the formatted changelog
changelog := repo.generateChangelog(releaseData.TagName, releaseData.Published, releaseData.HtmlUrl, releaseData.Body)
changelog, err := repo.generateChangelog(releaseData.TagName, releaseData.Published, releaseData.HtmlUrl, releaseData.Body)
if err != nil {
fmt.Fprintln(os.Stderr, "Error generating changelog:", err)
os.Exit(1)
}
fmt.Println(changelog)
}

Write Preview
Loading…
Cancel
Save