mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-15 22:39:03 +08:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 OpenIM-Gordon
|
6f49519053 | ||
|
OpenIM-Gordon
|
3ecde781bb |
@@ -1,65 +0,0 @@
|
|||||||
name: Cleanup After Milestone PRs Merged
|
|
||||||
|
|
||||||
on:
|
|
||||||
pull_request:
|
|
||||||
types:
|
|
||||||
- closed
|
|
||||||
|
|
||||||
jobs:
|
|
||||||
handle_pr:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
|
|
||||||
steps:
|
|
||||||
- name: Checkout repository
|
|
||||||
uses: actions/checkout@v4.2.0
|
|
||||||
|
|
||||||
- name: Get the PR title and extract PR numbers
|
|
||||||
id: extract_pr_numbers
|
|
||||||
run: |
|
|
||||||
# Get the PR title
|
|
||||||
PR_TITLE="${{ github.event.pull_request.title }}"
|
|
||||||
|
|
||||||
echo "PR Title: $PR_TITLE"
|
|
||||||
|
|
||||||
# Extract PR numbers from the title
|
|
||||||
PR_NUMBERS=$(echo "$PR_TITLE" | grep -oE "#[0-9]+" | tr -d '#' | tr '\n' ' ')
|
|
||||||
echo "Extracted PR Numbers: $PR_NUMBERS"
|
|
||||||
|
|
||||||
# Save PR numbers to a file
|
|
||||||
echo "$PR_NUMBERS" > pr_numbers.txt
|
|
||||||
echo "Saved PR Numbers to pr_numbers.txt"
|
|
||||||
|
|
||||||
# Check if the title matches a specific pattern
|
|
||||||
if echo "$PR_TITLE" | grep -qE "^deps: Merge( #[0-9]+)+ PRs into .+"; then
|
|
||||||
echo "proceed=true" >> $GITHUB_OUTPUT
|
|
||||||
else
|
|
||||||
echo "proceed=false" >> $GITHUB_OUTPUT
|
|
||||||
fi
|
|
||||||
|
|
||||||
- name: Use extracted PR numbers and label PRs
|
|
||||||
if: (steps.extract_pr_numbers.outputs.proceed == 'true' || contains(github.event.pull_request.labels.*.name, 'milestone-merge')) && github.event.pull_request.merged == true
|
|
||||||
run: |
|
|
||||||
# Read the previously saved PR numbers
|
|
||||||
PR_NUMBERS=$(cat pr_numbers.txt)
|
|
||||||
echo "Using extracted PR Numbers: $PR_NUMBERS"
|
|
||||||
|
|
||||||
# Loop through each PR number and add label
|
|
||||||
for PR_NUMBER in $PR_NUMBERS; do
|
|
||||||
echo "Adding 'cherry-picked' label to PR #$PR_NUMBER"
|
|
||||||
curl -X POST \
|
|
||||||
-H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
https://api.github.com/repos/${{ github.repository }}/issues/$PR_NUMBER/labels \
|
|
||||||
-d '{"labels":["cherry-picked"]}'
|
|
||||||
done
|
|
||||||
env:
|
|
||||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
|
||||||
|
|
||||||
- name: Delete branch after PR close
|
|
||||||
if: steps.extract_pr_numbers.outputs.proceed == 'true' || contains(github.event.pull_request.labels.*.name, 'milestone-merge')
|
|
||||||
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 }}
|
|
||||||
@@ -2,7 +2,11 @@ name: Go Build Test
|
|||||||
|
|
||||||
on:
|
on:
|
||||||
push:
|
push:
|
||||||
|
branches:
|
||||||
|
- main
|
||||||
pull_request:
|
pull_request:
|
||||||
|
branches:
|
||||||
|
- main
|
||||||
paths-ignore:
|
paths-ignore:
|
||||||
- '**/*.md'
|
- '**/*.md'
|
||||||
|
|
||||||
|
|||||||
@@ -1,218 +0,0 @@
|
|||||||
name: Create Pre-Release PR from Milestone
|
|
||||||
|
|
||||||
permissions:
|
|
||||||
contents: write
|
|
||||||
pull-requests: write
|
|
||||||
issues: write
|
|
||||||
|
|
||||||
on:
|
|
||||||
workflow_dispatch:
|
|
||||||
inputs:
|
|
||||||
milestone_name:
|
|
||||||
description: 'Milestone name to collect closed PRs from'
|
|
||||||
required: true
|
|
||||||
default: 'v3.8.2'
|
|
||||||
target_branch:
|
|
||||||
description: 'Target branch to merge the consolidated PR'
|
|
||||||
required: true
|
|
||||||
default: 'pre-release-v3.8.2'
|
|
||||||
|
|
||||||
env:
|
|
||||||
MILESTONE_NAME: ${{ github.event.inputs.milestone_name || 'v3.8.2' }}
|
|
||||||
TARGET_BRANCH: ${{ github.event.inputs.target_branch || 'pre-release-v3.8.2' }}
|
|
||||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
|
||||||
BOT_TOKEN: ${{ secrets.BOT_TOKEN }}
|
|
||||||
LABEL_NAME: cherry-picked
|
|
||||||
TEMP_DIR: /tmp # Using /tmp as the temporary directory
|
|
||||||
|
|
||||||
jobs:
|
|
||||||
cherry_pick_milestone_prs:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
steps:
|
|
||||||
- name: Setup temp directory
|
|
||||||
run: |
|
|
||||||
# Create the temporary directory and initialize necessary files
|
|
||||||
mkdir -p ${{ env.TEMP_DIR }}
|
|
||||||
touch ${{ env.TEMP_DIR }}/pr_numbers.txt
|
|
||||||
touch ${{ env.TEMP_DIR }}/commit_hashes.txt
|
|
||||||
touch ${{ env.TEMP_DIR }}/pr_title.txt
|
|
||||||
touch ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
touch ${{ env.TEMP_DIR }}/created_pr_number.txt
|
|
||||||
|
|
||||||
- name: Checkout repository
|
|
||||||
uses: actions/checkout@v4
|
|
||||||
with:
|
|
||||||
fetch-depth: 0
|
|
||||||
token: ${{ secrets.BOT_TOKEN }}
|
|
||||||
|
|
||||||
- name: Setup Git User for OpenIM-Robot
|
|
||||||
run: |
|
|
||||||
# Set up Git credentials for the bot
|
|
||||||
git config --global user.email "OpenIM-Robot@users.noreply.github.com"
|
|
||||||
git config --global user.name "OpenIM-Robot"
|
|
||||||
|
|
||||||
- name: Fetch Milestone ID and Filter PR Numbers
|
|
||||||
env:
|
|
||||||
MILESTONE_NAME: ${{ env.MILESTONE_NAME }}
|
|
||||||
run: |
|
|
||||||
# Fetch milestone details and extract milestone ID
|
|
||||||
milestones=$(curl -s -H "Authorization: token $BOT_TOKEN" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
"https://api.github.com/repos/${{ github.repository }}/milestones")
|
|
||||||
milestone_id=$(echo "$milestones" | grep -B3 "\"title\": \"$MILESTONE_NAME\"" | grep '"number":' | head -n1 | grep -o '[0-9]\+')
|
|
||||||
if [ -z "$milestone_id" ]; then
|
|
||||||
echo "Milestone '$MILESTONE_NAME' not found. Exiting."
|
|
||||||
exit 1
|
|
||||||
fi
|
|
||||||
echo "Milestone ID: $milestone_id"
|
|
||||||
echo "MILESTONE_ID=$milestone_id" >> $GITHUB_ENV
|
|
||||||
|
|
||||||
# Fetch issues for the milestone
|
|
||||||
issues=$(curl -s -H "Authorization: token $BOT_TOKEN" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
"https://api.github.com/repos/${{ github.repository }}/issues?milestone=$milestone_id&state=closed&per_page=100")
|
|
||||||
|
|
||||||
> ${{ env.TEMP_DIR }}/pr_numbers.txt
|
|
||||||
|
|
||||||
# Filter PRs that do not have the 'cherry-picked' label
|
|
||||||
for pr_number in $(echo "$issues" | jq -r '.[] | select(.pull_request != null) | .number'); do
|
|
||||||
labels=$(curl -s -H "Authorization: token $BOT_TOKEN" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
"https://api.github.com/repos/${{ github.repository }}/issues/$pr_number/labels" | jq -r '.[].name')
|
|
||||||
|
|
||||||
if ! echo "$labels" | grep -q "${LABEL_NAME}"; then
|
|
||||||
echo "PR #$pr_number does not have the 'cherry-picked' label. Adding to the list."
|
|
||||||
echo "$pr_number" >> ${{ env.TEMP_DIR }}/pr_numbers.txt
|
|
||||||
else
|
|
||||||
echo "PR #$pr_number already has the 'cherry-picked' label. Skipping."
|
|
||||||
fi
|
|
||||||
done
|
|
||||||
|
|
||||||
# Sort the filtered PR numbers
|
|
||||||
sort -n ${{ env.TEMP_DIR }}/pr_numbers.txt -o ${{ env.TEMP_DIR }}/pr_numbers.txt
|
|
||||||
|
|
||||||
echo "Filtered and sorted PR numbers:"
|
|
||||||
cat ${{ env.TEMP_DIR }}/pr_numbers.txt || echo "No closed PR numbers found for milestone."
|
|
||||||
|
|
||||||
- name: Fetch Merge Commits for PRs and Generate Title and Body
|
|
||||||
run: |
|
|
||||||
# Ensure the files are initialized
|
|
||||||
> ${{ env.TEMP_DIR }}/commit_hashes.txt
|
|
||||||
> ${{ env.TEMP_DIR }}/pr_title.txt
|
|
||||||
> ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
|
|
||||||
# Write description to the PR body
|
|
||||||
echo "### Description:" >> ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
echo "Merging PRs from milestone \`$MILESTONE_NAME\` into target branch \`$TARGET_BRANCH\`." >> ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
echo "" >> ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
echo "### Need Merge PRs:" >> ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
|
|
||||||
pr_numbers_in_title=""
|
|
||||||
|
|
||||||
# Process sorted PR numbers and generate commit hashes
|
|
||||||
for pr_number in $(cat ${{ env.TEMP_DIR }}/pr_numbers.txt); do
|
|
||||||
echo "Processing PR #$pr_number"
|
|
||||||
pr_details=$(curl -s -H "Authorization: token $BOT_TOKEN" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
"https://api.github.com/repos/${{ github.repository }}/pulls/$pr_number")
|
|
||||||
pr_title=$(echo "$pr_details" | jq -r '.title')
|
|
||||||
merge_commit=$(echo "$pr_details" | jq -r '.merge_commit_sha')
|
|
||||||
short_commit_hash=$(echo "$merge_commit" | cut -c 1-7)
|
|
||||||
|
|
||||||
# Append PR details to the body
|
|
||||||
echo "- $pr_title: (#$pr_number) ($short_commit_hash)" >> ${{ env.TEMP_DIR }}/pr_body.txt
|
|
||||||
|
|
||||||
if [ "$merge_commit" != "null" ];then
|
|
||||||
echo "$merge_commit" >> ${{ env.TEMP_DIR }}/commit_hashes.txt
|
|
||||||
echo "#$pr_number" >> ${{ env.TEMP_DIR }}/pr_title.txt
|
|
||||||
pr_numbers_in_title="$pr_numbers_in_title #$pr_number"
|
|
||||||
fi
|
|
||||||
done
|
|
||||||
|
|
||||||
commit_hashes=$(cat ${{ env.TEMP_DIR }}/commit_hashes.txt | tr '\n' ' ')
|
|
||||||
first_commit_hash=$(head -n 1 ${{ env.TEMP_DIR }}/commit_hashes.txt)
|
|
||||||
cherry_pick_branch="cherry-pick-${first_commit_hash:0:7}"
|
|
||||||
echo "COMMIT_HASHES=$commit_hashes" >> $GITHUB_ENV
|
|
||||||
echo "CHERRY_PICK_BRANCH=$cherry_pick_branch" >> $GITHUB_ENV
|
|
||||||
echo "pr_numbers_in_title=$pr_numbers_in_title" >> $GITHUB_ENV
|
|
||||||
|
|
||||||
- name: Pull and Cherry-pick Commits, Then Push
|
|
||||||
env:
|
|
||||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
|
||||||
BOT_TOKEN: ${{ secrets.BOT_TOKEN }}
|
|
||||||
run: |
|
|
||||||
# Fetch and pull the latest changes from the target branch
|
|
||||||
git fetch origin
|
|
||||||
git checkout $TARGET_BRANCH
|
|
||||||
git pull origin $TARGET_BRANCH
|
|
||||||
|
|
||||||
# Create a new branch for cherry-picking
|
|
||||||
git checkout -b $CHERRY_PICK_BRANCH
|
|
||||||
|
|
||||||
# Cherry-pick the commits and handle conflicts
|
|
||||||
for commit_hash in $COMMIT_HASHES; do
|
|
||||||
echo "Attempting to cherry-pick commit $commit_hash"
|
|
||||||
if ! git cherry-pick "$commit_hash" --strategy=recursive -X theirs; then
|
|
||||||
echo "Conflict detected for $commit_hash. Resolving with incoming changes."
|
|
||||||
conflict_files=$(git diff --name-only --diff-filter=U)
|
|
||||||
echo "Conflicting files:"
|
|
||||||
echo "$conflict_files"
|
|
||||||
|
|
||||||
for file in $conflict_files; do
|
|
||||||
if [ -f "$file" ]; then
|
|
||||||
echo "Resolving conflict for $file"
|
|
||||||
git add "$file"
|
|
||||||
else
|
|
||||||
echo "File $file has been deleted. Skipping."
|
|
||||||
git rm "$file"
|
|
||||||
fi
|
|
||||||
done
|
|
||||||
|
|
||||||
echo "Conflicts resolved. Continuing cherry-pick."
|
|
||||||
git cherry-pick --continue
|
|
||||||
else
|
|
||||||
echo "Cherry-pick successful for commit $commit_hash."
|
|
||||||
fi
|
|
||||||
done
|
|
||||||
|
|
||||||
# Push the cherry-pick branch to the repository
|
|
||||||
git remote set-url origin "https://${BOT_TOKEN}@github.com/${{ github.repository }}.git"
|
|
||||||
git push origin $CHERRY_PICK_BRANCH --force
|
|
||||||
|
|
||||||
- name: Create Pull Request
|
|
||||||
run: |
|
|
||||||
# Prepare and create the PR
|
|
||||||
pr_title="deps: Merge ${{ env.pr_numbers_in_title }} PRs into $TARGET_BRANCH"
|
|
||||||
pr_body=$(cat ${{ env.TEMP_DIR }}/pr_body.txt)
|
|
||||||
|
|
||||||
echo "Prepared PR title:"
|
|
||||||
echo "$pr_title"
|
|
||||||
echo "Prepared PR body:"
|
|
||||||
echo "$pr_body"
|
|
||||||
|
|
||||||
# Create the PR using the GitHub API
|
|
||||||
response=$(curl -s -X POST -H "Authorization: token $BOT_TOKEN" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
https://api.github.com/repos/${{ github.repository }}/pulls \
|
|
||||||
-d "$(jq -n --arg title "$pr_title" \
|
|
||||||
--arg head "$CHERRY_PICK_BRANCH" \
|
|
||||||
--arg base "$TARGET_BRANCH" \
|
|
||||||
--arg body "$pr_body" \
|
|
||||||
'{title: $title, head: $head, base: $base, body: $body}')")
|
|
||||||
|
|
||||||
pr_number=$(echo "$response" | jq -r '.number')
|
|
||||||
echo "$pr_number" > ${{ env.TEMP_DIR }}/created_pr_number.txt
|
|
||||||
echo "Created PR #$pr_number"
|
|
||||||
|
|
||||||
- name: Add Label to Created Pull Request
|
|
||||||
run: |
|
|
||||||
# Add 'milestone-merge' label to the created PR
|
|
||||||
pr_number=$(cat ${{ env.TEMP_DIR }}/created_pr_number.txt)
|
|
||||||
echo "Adding label to PR #$pr_number"
|
|
||||||
|
|
||||||
curl -s -X POST -H "Authorization: token $GITHUB_TOKEN" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
-d '{"labels": ["milestone-merge"]}' \
|
|
||||||
"https://api.github.com/repos/${{ github.repository }}/issues/$pr_number/labels"
|
|
||||||
|
|
||||||
echo "Added 'milestone-merge' label to PR #$pr_number."
|
|
||||||
@@ -0,0 +1,62 @@
|
|||||||
|
# Version logging for OpenIM
|
||||||
|
|
||||||
|
<!-- BEGIN MUNGE: GENERATED_TOC -->
|
||||||
|
|
||||||
|
<!-- END MUNGE: GENERATED_TOC -->
|
||||||
|
|
||||||
|
{{ if .Versions -}}
|
||||||
|
<a name="unreleased"></a>
|
||||||
|
## [Unreleased]
|
||||||
|
|
||||||
|
{{ if .Unreleased.CommitGroups -}}
|
||||||
|
{{ range .Unreleased.CommitGroups -}}
|
||||||
|
### {{ .Title }}
|
||||||
|
{{ range .Commits -}}
|
||||||
|
- {{ if .Scope }}**{{ .Scope }}:** {{ end }}{{ .Subject }}
|
||||||
|
{{ end }}
|
||||||
|
{{ end -}}
|
||||||
|
{{ end -}}
|
||||||
|
{{ end -}}
|
||||||
|
|
||||||
|
{{ range .Versions }}
|
||||||
|
<a name="{{ .Tag.Name }}"></a>
|
||||||
|
## {{ if .Tag.Previous }}[{{ .Tag.Name }}]{{ else }}{{ .Tag.Name }}{{ end }} - {{ datetime "2006-01-02" .Tag.Date }}
|
||||||
|
{{ range .CommitGroups -}}
|
||||||
|
### {{ .Title }}
|
||||||
|
{{ range .Commits -}}
|
||||||
|
- {{ if .Scope }}**{{ .Scope }}:** {{ end }}{{ .Subject }}
|
||||||
|
{{ end }}
|
||||||
|
{{ end -}}
|
||||||
|
|
||||||
|
{{- if .RevertCommits -}}
|
||||||
|
### Reverts
|
||||||
|
{{ range .RevertCommits -}}
|
||||||
|
- {{ .Revert.Header }}
|
||||||
|
{{ end }}
|
||||||
|
{{ end -}}
|
||||||
|
|
||||||
|
{{- if .MergeCommits -}}
|
||||||
|
### Pull Requests
|
||||||
|
{{ range .MergeCommits -}}
|
||||||
|
- {{ .Header }}
|
||||||
|
{{ end }}
|
||||||
|
{{ end -}}
|
||||||
|
|
||||||
|
{{- if .NoteGroups -}}
|
||||||
|
{{ range .NoteGroups -}}
|
||||||
|
### {{ .Title }}
|
||||||
|
{{ range .Notes }}
|
||||||
|
{{ .Body }}
|
||||||
|
{{ end }}
|
||||||
|
{{ end -}}
|
||||||
|
{{ end -}}
|
||||||
|
{{ end -}}
|
||||||
|
|
||||||
|
{{- if .Versions }}
|
||||||
|
[Unreleased]: {{ .Info.RepositoryURL }}/compare/{{ $latest := index .Versions 0 }}{{ $latest.Tag.Name }}...HEAD
|
||||||
|
{{ range .Versions -}}
|
||||||
|
{{ if .Tag.Previous -}}
|
||||||
|
[{{ .Tag.Name }}]: {{ $.Info.RepositoryURL }}/compare/{{ .Tag.Previous.Name }}...{{ .Tag.Name }}
|
||||||
|
{{ end -}}
|
||||||
|
{{ end -}}
|
||||||
|
{{ end -}}
|
||||||
@@ -0,0 +1,67 @@
|
|||||||
|
bin: git
|
||||||
|
style: github
|
||||||
|
template: CHANGELOG.tpl.md
|
||||||
|
info:
|
||||||
|
title: CHANGELOG
|
||||||
|
repository_url: https://github.com/openimsdk/open-im-server
|
||||||
|
options:
|
||||||
|
tag_filter_pattern: '^v'
|
||||||
|
sort: "date"
|
||||||
|
|
||||||
|
commits:
|
||||||
|
filters:
|
||||||
|
Type:
|
||||||
|
- feat
|
||||||
|
- fix
|
||||||
|
- perf
|
||||||
|
- refactor
|
||||||
|
- docs
|
||||||
|
- test
|
||||||
|
- chore
|
||||||
|
- ci
|
||||||
|
- build
|
||||||
|
sort_by: Scope
|
||||||
|
|
||||||
|
commit_groups:
|
||||||
|
group_by: Type
|
||||||
|
sort_by: Title
|
||||||
|
title_order:
|
||||||
|
- feat
|
||||||
|
- fix
|
||||||
|
- perf
|
||||||
|
- refactor
|
||||||
|
- docs
|
||||||
|
- test
|
||||||
|
- chore
|
||||||
|
- ci
|
||||||
|
- build
|
||||||
|
title_maps:
|
||||||
|
feat: Features
|
||||||
|
|
||||||
|
header:
|
||||||
|
pattern: "<regexp>"
|
||||||
|
pattern_maps:
|
||||||
|
- PropName
|
||||||
|
|
||||||
|
issues:
|
||||||
|
prefix:
|
||||||
|
- #
|
||||||
|
|
||||||
|
refs:
|
||||||
|
actions:
|
||||||
|
- Closes
|
||||||
|
- Fixes
|
||||||
|
|
||||||
|
merges:
|
||||||
|
pattern: "^Merge branch '(\\w+)'$"
|
||||||
|
pattern_maps:
|
||||||
|
- Source
|
||||||
|
|
||||||
|
reverts:
|
||||||
|
pattern: "^Revert \"([\\s\\S]*)\"$"
|
||||||
|
pattern_maps:
|
||||||
|
- Header
|
||||||
|
|
||||||
|
notes:
|
||||||
|
keywords:
|
||||||
|
- BREAKING CHANGE
|
||||||
@@ -53,8 +53,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-cmdutils
|
- binary: openim-cmdutils
|
||||||
id: openim-cmdutils
|
id: openim-cmdutils
|
||||||
@@ -64,8 +71,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-crontask
|
- binary: openim-crontask
|
||||||
id: openim-crontask
|
id: openim-crontask
|
||||||
@@ -75,8 +89,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-msggateway
|
- binary: openim-msggateway
|
||||||
id: openim-msggateway
|
id: openim-msggateway
|
||||||
@@ -86,8 +107,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-msgtransfer
|
- binary: openim-msgtransfer
|
||||||
id: openim-msgtransfer
|
id: openim-msgtransfer
|
||||||
@@ -97,8 +125,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-push
|
- binary: openim-push
|
||||||
id: openim-push
|
id: openim-push
|
||||||
@@ -108,8 +143,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-auth
|
- binary: openim-rpc-auth
|
||||||
id: openim-rpc-auth
|
id: openim-rpc-auth
|
||||||
@@ -119,8 +161,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-conversation
|
- binary: openim-rpc-conversation
|
||||||
id: openim-rpc-conversation
|
id: openim-rpc-conversation
|
||||||
@@ -130,8 +179,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-friend
|
- binary: openim-rpc-friend
|
||||||
id: openim-rpc-friend
|
id: openim-rpc-friend
|
||||||
@@ -141,8 +197,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-group
|
- binary: openim-rpc-group
|
||||||
id: openim-rpc-group
|
id: openim-rpc-group
|
||||||
@@ -152,8 +215,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-msg
|
- binary: openim-rpc-msg
|
||||||
id: openim-rpc-msg
|
id: openim-rpc-msg
|
||||||
@@ -163,8 +233,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-third
|
- binary: openim-rpc-third
|
||||||
id: openim-rpc-third
|
id: openim-rpc-third
|
||||||
@@ -174,8 +251,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
- binary: openim-rpc-user
|
- binary: openim-rpc-user
|
||||||
id: openim-rpc-user
|
id: openim-rpc-user
|
||||||
@@ -185,8 +269,15 @@ builds:
|
|||||||
- windows
|
- windows
|
||||||
- linux
|
- linux
|
||||||
goarch:
|
goarch:
|
||||||
|
- s390x
|
||||||
|
- mips64
|
||||||
|
- mips64le
|
||||||
- amd64
|
- amd64
|
||||||
|
- ppc64le
|
||||||
- arm64
|
- arm64
|
||||||
|
goarm:
|
||||||
|
- "6"
|
||||||
|
- "7"
|
||||||
|
|
||||||
|
|
||||||
# TODO：Need a script, such as the init - release to help binary to find the right directory
|
# TODO：Need a script, such as the init - release to help binary to find the right directory
|
||||||
|
|||||||
@@ -5,4 +5,9 @@ etcd:
|
|||||||
username: ''
|
username: ''
|
||||||
password: ''
|
password: ''
|
||||||
|
|
||||||
|
zookeeper:
|
||||||
|
schema: openim
|
||||||
|
address: [ localhost:12181 ]
|
||||||
|
username: ''
|
||||||
|
password: ''
|
||||||
|
|
||||||
|
|||||||
+20
-3
@@ -1,3 +1,20 @@
|
|||||||
|
# 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.
|
||||||
|
|
||||||
|
# Determines if a message should be sent. If set to false, it triggers a silent sync without a message. If true, it requires triggering a conversation.
|
||||||
|
# For rpc notification, send twice: once as a message and once as a notification.
|
||||||
|
# The options field 'isNotification' indicates if it's a notification.
|
||||||
groupCreated:
|
groupCreated:
|
||||||
isSendMsg: true
|
isSendMsg: true
|
||||||
# Reliability level of the message sending.
|
# Reliability level of the message sending.
|
||||||
@@ -292,9 +309,9 @@ userInfoUpdated:
|
|||||||
unreadCount: false
|
unreadCount: false
|
||||||
offlinePush:
|
offlinePush:
|
||||||
enable: true
|
enable: true
|
||||||
title: userInfo updated
|
title: Remove a blocked user
|
||||||
desc: userInfo updated
|
desc: Remove a blocked user
|
||||||
ext: userInfo updated
|
ext: Remove a blocked user
|
||||||
|
|
||||||
userStatusChanged:
|
userStatusChanged:
|
||||||
isSendMsg: false
|
isSendMsg: false
|
||||||
|
|||||||
+1
-14
@@ -13,17 +13,4 @@ rpcRegisterName:
|
|||||||
imAdminUserID: [ imAdmin ]
|
imAdminUserID: [ imAdmin ]
|
||||||
|
|
||||||
# 1: For Android, iOS, Windows, Mac, and web platforms, only one instance can be online at a time
|
# 1: For Android, iOS, Windows, Mac, and web platforms, only one instance can be online at a time
|
||||||
multiLogin:
|
multiLoginPolicy: 1
|
||||||
policy: 1
|
|
||||||
maxNumOneEnd: 30
|
|
||||||
customizeLoginNum:
|
|
||||||
ios: 1
|
|
||||||
android: 1
|
|
||||||
windows: 1
|
|
||||||
osx: 1
|
|
||||||
web: 1
|
|
||||||
miniWeb: 1
|
|
||||||
linux: 1
|
|
||||||
aPad: 1
|
|
||||||
iPad: 1
|
|
||||||
admin: 1
|
|
||||||
|
|||||||
+13
-1
@@ -43,6 +43,19 @@ services:
|
|||||||
networks:
|
networks:
|
||||||
- openim
|
- openim
|
||||||
|
|
||||||
|
zookeeper:
|
||||||
|
image: "${ZOOKEEPER_IMAGE}"
|
||||||
|
container_name: zookeeper
|
||||||
|
ports:
|
||||||
|
- "12181:2181"
|
||||||
|
environment:
|
||||||
|
#JVMFLAGS: "-Xms32m -Xmx128m"
|
||||||
|
TZ: "Asia/Shanghai"
|
||||||
|
ALLOW_ANONYMOUS_LOGIN: "yes"
|
||||||
|
restart: always
|
||||||
|
networks:
|
||||||
|
- openim
|
||||||
|
|
||||||
etcd:
|
etcd:
|
||||||
image: "${ETCD_IMAGE}"
|
image: "${ETCD_IMAGE}"
|
||||||
container_name: etcd
|
container_name: etcd
|
||||||
@@ -129,7 +142,6 @@ services:
|
|||||||
# image: ${PROMETHEUS_IMAGE}
|
# image: ${PROMETHEUS_IMAGE}
|
||||||
# container_name: prometheus
|
# container_name: prometheus
|
||||||
# restart: always
|
# restart: always
|
||||||
# user: root
|
|
||||||
# volumes:
|
# volumes:
|
||||||
# - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
|
# - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
|
||||||
# - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
|
# - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ require (
|
|||||||
github.com/gorilla/websocket v1.5.1
|
github.com/gorilla/websocket v1.5.1
|
||||||
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
|
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
|
||||||
github.com/mitchellh/mapstructure v1.5.0
|
github.com/mitchellh/mapstructure v1.5.0
|
||||||
github.com/openimsdk/protocol v0.0.72-pre-release-v3.8.2-alpha.5
|
github.com/openimsdk/protocol v0.0.72
|
||||||
github.com/openimsdk/tools v0.0.50-alpha.16
|
github.com/openimsdk/tools v0.0.50-alpha.16
|
||||||
github.com/pkg/errors v0.9.1 // indirect
|
github.com/pkg/errors v0.9.1 // indirect
|
||||||
github.com/prometheus/client_golang v1.18.0
|
github.com/prometheus/client_golang v1.18.0
|
||||||
|
|||||||
@@ -319,8 +319,8 @@ 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/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 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
|
||||||
github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
|
github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
|
||||||
github.com/openimsdk/protocol v0.0.72-pre-release-v3.8.2-alpha.5 h1:b0JAuBhzIYirHeXp7asB04bE1q+KhU3dpAaAroc/Am0=
|
github.com/openimsdk/protocol v0.0.72 h1:K+vslwaR7lDXyBzb07UuEQITaqsgighz7NyXVIWsu6A=
|
||||||
github.com/openimsdk/protocol v0.0.72-pre-release-v3.8.2-alpha.5/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
|
github.com/openimsdk/protocol v0.0.72/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
|
||||||
github.com/openimsdk/tools v0.0.50-alpha.16 h1:bC1AQvJMuOHtZm8LZRvN8L5mH1Ws2VYdL+TLTs1iGSc=
|
github.com/openimsdk/tools v0.0.50-alpha.16 h1:bC1AQvJMuOHtZm8LZRvN8L5mH1Ws2VYdL+TLTs1iGSc=
|
||||||
github.com/openimsdk/tools v0.0.50-alpha.16/go.mod h1:h1cYmfyaVtgFbKmb1Cfsl8XwUOMTt8ubVUQrdGtsUh4=
|
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 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
|
||||||
|
|||||||
@@ -90,19 +90,6 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
|
|||||||
if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil {
|
if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil {
|
||||||
log.ZError(ctx, "update user online status", err)
|
log.ZError(ctx, "update user online status", err)
|
||||||
}
|
}
|
||||||
for _, ss := range req.Status {
|
|
||||||
for _, online := range ss.Online {
|
|
||||||
client, _, _ := ws.clients.Get(ss.UserID, int(online))
|
|
||||||
back := false
|
|
||||||
if len(client) > 0 {
|
|
||||||
back = client[0].IsBackground
|
|
||||||
}
|
|
||||||
ws.webhookAfterUserOnline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOnline, ss.UserID, int(online), back, ss.ConnID)
|
|
||||||
}
|
|
||||||
for _, offline := range ss.Offline {
|
|
||||||
ws.webhookAfterUserOffline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOffline, ss.UserID, int(offline), ss.ConnID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := 0; i < concurrent; i++ {
|
for i := 0; i < concurrent; i++ {
|
||||||
|
|||||||
@@ -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 msggateway
|
package msggateway
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -198,6 +212,7 @@ func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *C
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
wg := errgroup.Group{}
|
wg := errgroup.Group{}
|
||||||
wg.SetLimit(concurrentRequest)
|
wg.SetLimit(concurrentRequest)
|
||||||
|
|
||||||
@@ -306,32 +321,8 @@ func (ws *WsServer) KickUserConn(client *Client) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Client, newClient *Client) {
|
func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Client, newClient *Client) {
|
||||||
kickTokenFunc := func(kickClients []*Client) {
|
switch ws.msgGatewayConfig.Share.MultiLoginPolicy {
|
||||||
var kickTokens []string
|
|
||||||
ws.clients.DeleteClients(newClient.UserID, kickClients)
|
|
||||||
for _, c := range kickClients {
|
|
||||||
kickTokens = append(kickTokens, c.token)
|
|
||||||
err := c.KickOnlineMessage()
|
|
||||||
if err != nil {
|
|
||||||
log.ZWarn(c.ctx, "KickOnlineMessage", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
ctx := mcontext.WithMustInfoCtx(
|
|
||||||
[]string{newClient.ctx.GetOperationID(), newClient.ctx.GetUserID(),
|
|
||||||
constant.PlatformIDToName(newClient.PlatformID), newClient.ctx.GetConnID()},
|
|
||||||
)
|
|
||||||
if _, err := ws.authClient.KickTokens(ctx, kickTokens); err != nil {
|
|
||||||
log.ZWarn(newClient.ctx, "kickTokens err", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
switch ws.msgGatewayConfig.Share.MultiLogin.Policy {
|
|
||||||
case constant.DefalutNotKick:
|
case constant.DefalutNotKick:
|
||||||
case constant.WebAndOther:
|
|
||||||
if constant.PlatformIDToClass(newClient.PlatformID) == constant.WebPlatformStr {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
fallthrough
|
|
||||||
case constant.PCAndOther:
|
case constant.PCAndOther:
|
||||||
if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC {
|
if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC {
|
||||||
return
|
return
|
||||||
@@ -356,35 +347,6 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
|
|||||||
log.ZWarn(newClient.ctx, "InvalidateToken err", err, "userID", newClient.UserID,
|
log.ZWarn(newClient.ctx, "InvalidateToken err", err, "userID", newClient.UserID,
|
||||||
"platformID", newClient.PlatformID)
|
"platformID", newClient.PlatformID)
|
||||||
}
|
}
|
||||||
case constant.PcMobileAndWeb:
|
|
||||||
clients, ok := ws.clients.GetAll(newClient.UserID)
|
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
var (
|
|
||||||
kickClients []*Client
|
|
||||||
)
|
|
||||||
for _, client := range clients {
|
|
||||||
if constant.PlatformIDToClass(client.PlatformID) == constant.PlatformIDToClass(newClient.PlatformID) {
|
|
||||||
kickClients = append(kickClients, client)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
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
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -128,7 +128,6 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
|
|||||||
|
|
||||||
go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH)
|
go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH)
|
||||||
go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH)
|
go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH)
|
||||||
go m.historyCH.HandleUserHasReadSeqMessages(m.ctx)
|
|
||||||
err := m.historyCH.redisMessageBatches.Start()
|
err := m.historyCH.redisMessageBatches.Start()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -158,14 +157,12 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
|
|||||||
// graceful close kafka client.
|
// graceful close kafka client.
|
||||||
m.cancel()
|
m.cancel()
|
||||||
m.historyCH.redisMessageBatches.Close()
|
m.historyCH.redisMessageBatches.Close()
|
||||||
m.historyCH.Close()
|
|
||||||
m.historyCH.historyConsumerGroup.Close()
|
m.historyCH.historyConsumerGroup.Close()
|
||||||
m.historyMongoCH.historyConsumerGroup.Close()
|
m.historyMongoCH.historyConsumerGroup.Close()
|
||||||
return nil
|
return nil
|
||||||
case <-netDone:
|
case <-netDone:
|
||||||
m.cancel()
|
m.cancel()
|
||||||
m.historyCH.redisMessageBatches.Close()
|
m.historyCH.redisMessageBatches.Close()
|
||||||
m.historyCH.Close()
|
|
||||||
m.historyCH.historyConsumerGroup.Close()
|
m.historyCH.historyConsumerGroup.Close()
|
||||||
m.historyMongoCH.historyConsumerGroup.Close()
|
m.historyMongoCH.historyConsumerGroup.Close()
|
||||||
close(netDone)
|
close(netDone)
|
||||||
|
|||||||
@@ -18,10 +18,8 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/IBM/sarama"
|
"github.com/IBM/sarama"
|
||||||
@@ -42,12 +40,11 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
size = 500
|
size = 500
|
||||||
mainDataBuffer = 500
|
mainDataBuffer = 500
|
||||||
subChanBuffer = 50
|
subChanBuffer = 50
|
||||||
worker = 50
|
worker = 50
|
||||||
interval = 100 * time.Millisecond
|
interval = 100 * time.Millisecond
|
||||||
hasReadChanBuffer = 1000
|
|
||||||
)
|
)
|
||||||
|
|
||||||
type ContextMsg struct {
|
type ContextMsg struct {
|
||||||
@@ -55,23 +52,14 @@ type ContextMsg struct {
|
|||||||
ctx context.Context
|
ctx context.Context
|
||||||
}
|
}
|
||||||
|
|
||||||
// This structure is used for asynchronously writing the sender’s read sequence (seq) regarding a message into MongoDB.
|
|
||||||
// For example, if the sender sends a message with a seq of 10, then their own read seq for this conversation should be set to 10.
|
|
||||||
type userHasReadSeq struct {
|
|
||||||
conversationID string
|
|
||||||
userHasReadMap map[string]int64
|
|
||||||
}
|
|
||||||
|
|
||||||
type OnlineHistoryRedisConsumerHandler struct {
|
type OnlineHistoryRedisConsumerHandler struct {
|
||||||
historyConsumerGroup *kafka.MConsumerGroup
|
historyConsumerGroup *kafka.MConsumerGroup
|
||||||
|
|
||||||
redisMessageBatches *batcher.Batcher[sarama.ConsumerMessage]
|
redisMessageBatches *batcher.Batcher[sarama.ConsumerMessage]
|
||||||
|
|
||||||
msgTransferDatabase controller.MsgTransferDatabase
|
msgTransferDatabase controller.MsgTransferDatabase
|
||||||
conversationRpcClient *rpcclient.ConversationRpcClient
|
conversationRpcClient *rpcclient.ConversationRpcClient
|
||||||
groupRpcClient *rpcclient.GroupRpcClient
|
groupRpcClient *rpcclient.GroupRpcClient
|
||||||
conversationUserHasReadChan chan *userHasReadSeq
|
|
||||||
wg sync.WaitGroup
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase,
|
func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase,
|
||||||
@@ -82,8 +70,6 @@ func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database cont
|
|||||||
}
|
}
|
||||||
var och OnlineHistoryRedisConsumerHandler
|
var och OnlineHistoryRedisConsumerHandler
|
||||||
och.msgTransferDatabase = database
|
och.msgTransferDatabase = database
|
||||||
och.conversationUserHasReadChan = make(chan *userHasReadSeq, hasReadChanBuffer)
|
|
||||||
och.wg.Add(1)
|
|
||||||
|
|
||||||
b := batcher.New[sarama.ConsumerMessage](
|
b := batcher.New[sarama.ConsumerMessage](
|
||||||
batcher.WithSize(size),
|
batcher.WithSize(size),
|
||||||
@@ -129,25 +115,25 @@ func (och *OnlineHistoryRedisConsumerHandler) do(ctx context.Context, channelID
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context, msgs []*ContextMsg) {
|
func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context, msgs []*ContextMsg) {
|
||||||
|
type seqKey struct {
|
||||||
var conversationID string
|
conversationID string
|
||||||
var userSeqMap map[string]int64
|
userID string
|
||||||
|
}
|
||||||
|
var readSeq map[seqKey]int64
|
||||||
for _, msg := range msgs {
|
for _, msg := range msgs {
|
||||||
if msg.message.ContentType != constant.HasReadReceipt {
|
if msg.message.ContentType != constant.HasReadReceipt {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
var elem sdkws.NotificationElem
|
var elem sdkws.NotificationElem
|
||||||
if err := json.Unmarshal(msg.message.Content, &elem); err != nil {
|
if err := json.Unmarshal(msg.message.Content, &elem); err != nil {
|
||||||
log.ZWarn(ctx, "handlerConversationRead Unmarshal NotificationElem msg err", err, "msg", msg)
|
log.ZError(ctx, "handlerConversationRead Unmarshal NotificationElem msg err", err, "msg", msg)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
var tips sdkws.MarkAsReadTips
|
var tips sdkws.MarkAsReadTips
|
||||||
if err := json.Unmarshal([]byte(elem.Detail), &tips); err != nil {
|
if err := json.Unmarshal([]byte(elem.Detail), &tips); err != nil {
|
||||||
log.ZWarn(ctx, "handlerConversationRead Unmarshal MarkAsReadTips msg err", err, "msg", msg)
|
log.ZError(ctx, "handlerConversationRead Unmarshal MarkAsReadTips msg err", err, "msg", msg)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
//The conversation ID for each batch of messages processed by the batcher is the same.
|
|
||||||
conversationID = tips.ConversationID
|
|
||||||
if len(tips.Seqs) > 0 {
|
if len(tips.Seqs) > 0 {
|
||||||
for _, seq := range tips.Seqs {
|
for _, seq := range tips.Seqs {
|
||||||
if tips.HasReadSeq < seq {
|
if tips.HasReadSeq < seq {
|
||||||
@@ -160,25 +146,26 @@ func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context,
|
|||||||
if tips.HasReadSeq < 0 {
|
if tips.HasReadSeq < 0 {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if userSeqMap == nil {
|
if readSeq == nil {
|
||||||
userSeqMap = make(map[string]int64)
|
readSeq = make(map[seqKey]int64)
|
||||||
}
|
}
|
||||||
|
key := seqKey{
|
||||||
if userSeqMap[tips.MarkAsReadUserID] > tips.HasReadSeq {
|
conversationID: tips.ConversationID,
|
||||||
|
userID: tips.MarkAsReadUserID,
|
||||||
|
}
|
||||||
|
if readSeq[key] > tips.HasReadSeq {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
userSeqMap[tips.MarkAsReadUserID] = tips.HasReadSeq
|
readSeq[key] = tips.HasReadSeq
|
||||||
}
|
}
|
||||||
if userSeqMap == nil {
|
if readSeq == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if len(conversationID) == 0 {
|
for key, seq := range readSeq {
|
||||||
log.ZWarn(ctx, "conversation err", nil, "conversationID", conversationID)
|
if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, key.userID, key.conversationID, seq); err != nil {
|
||||||
|
log.ZError(ctx, "set read seq to db error", err, "userID", key.userID, "conversationID", key.conversationID, "seq", seq)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, conversationID, userSeqMap); err != nil {
|
|
||||||
log.ZWarn(ctx, "set read seq to db error", err, "conversationID", conversationID, "userSeqMap", userSeqMap)
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (och *OnlineHistoryRedisConsumerHandler) parseConsumerMessages(ctx context.Context, consumerMessages []*sarama.ConsumerMessage) []*ContextMsg {
|
func (och *OnlineHistoryRedisConsumerHandler) parseConsumerMessages(ctx context.Context, consumerMessages []*sarama.ConsumerMessage) []*ContextMsg {
|
||||||
@@ -263,21 +250,12 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key
|
|||||||
}
|
}
|
||||||
if len(storageMessageList) > 0 {
|
if len(storageMessageList) > 0 {
|
||||||
msg := storageMessageList[0]
|
msg := storageMessageList[0]
|
||||||
lastSeq, isNewConversation, userSeqMap, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
lastSeq, isNewConversation, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
||||||
if err != nil && !errors.Is(errs.Unwrap(err), redis.Nil) {
|
if err != nil && !errors.Is(errs.Unwrap(err), redis.Nil) {
|
||||||
log.ZWarn(ctx, "batch data insert to redis err", err, "storageMsgList", storageMessageList)
|
log.ZError(ctx, "batch data insert to redis err", err, "storageMsgList", storageMessageList)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
log.ZInfo(ctx, "BatchInsertChat2Cache end")
|
log.ZInfo(ctx, "BatchInsertChat2Cache end")
|
||||||
err = och.msgTransferDatabase.SetHasReadSeqs(ctx, conversationID, userSeqMap)
|
|
||||||
if err != nil {
|
|
||||||
log.ZWarn(ctx, "SetHasReadSeqs error", err, "userSeqMap", userSeqMap, "conversationID", conversationID)
|
|
||||||
prommetrics.SeqSetFailedCounter.Inc()
|
|
||||||
}
|
|
||||||
och.conversationUserHasReadChan <- &userHasReadSeq{
|
|
||||||
conversationID: conversationID,
|
|
||||||
userHasReadMap: userSeqMap,
|
|
||||||
}
|
|
||||||
|
|
||||||
if isNewConversation {
|
if isNewConversation {
|
||||||
switch msg.SessionType {
|
switch msg.SessionType {
|
||||||
@@ -330,7 +308,7 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
|
|||||||
storageMessageList = append(storageMessageList, msg.message)
|
storageMessageList = append(storageMessageList, msg.message)
|
||||||
}
|
}
|
||||||
if len(storageMessageList) > 0 {
|
if len(storageMessageList) > 0 {
|
||||||
lastSeq, _, _, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
lastSeq, _, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.ZError(ctx, "notification batch insert to redis error", err, "conversationID", conversationID,
|
log.ZError(ctx, "notification batch insert to redis error", err, "conversationID", conversationID,
|
||||||
"storageList", storageMessageList)
|
"storageList", storageMessageList)
|
||||||
@@ -345,21 +323,6 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
|
|||||||
och.toPushTopic(ctx, key, conversationID, storageList)
|
och.toPushTopic(ctx, key, conversationID, storageList)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
func (och *OnlineHistoryRedisConsumerHandler) HandleUserHasReadSeqMessages(ctx context.Context) {
|
|
||||||
defer och.wg.Done()
|
|
||||||
|
|
||||||
for msg := range och.conversationUserHasReadChan {
|
|
||||||
if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, msg.conversationID, msg.userHasReadMap); err != nil {
|
|
||||||
log.ZWarn(ctx, "set read seq to db error", err, "conversationID", msg.conversationID, "userSeqMap", msg.userHasReadMap)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
log.ZInfo(ctx, "Channel closed, exiting handleUserHasReadSeqMessages")
|
|
||||||
}
|
|
||||||
func (och *OnlineHistoryRedisConsumerHandler) Close() {
|
|
||||||
close(och.conversationUserHasReadChan)
|
|
||||||
och.wg.Wait()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, key, conversationID string, msgs []*ContextMsg) {
|
func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, key, conversationID string, msgs []*ContextMsg) {
|
||||||
for _, v := range msgs {
|
for _, v := range msgs {
|
||||||
|
|||||||
@@ -0,0 +1,29 @@
|
|||||||
|
package push
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/openimsdk/protocol/sdkws"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestName(t *testing.T) {
|
||||||
|
var c ConsumerHandler
|
||||||
|
c.readCh = make(chan *sdkws.MarkAsReadTips)
|
||||||
|
|
||||||
|
go c.loopRead()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
for i := 0; ; i++ {
|
||||||
|
seq := int64(i + 1)
|
||||||
|
if seq%3 == 0 {
|
||||||
|
seq = 1
|
||||||
|
}
|
||||||
|
c.readCh <- &sdkws.MarkAsReadTips{
|
||||||
|
ConversationID: "c100",
|
||||||
|
MarkAsReadUserID: "u100",
|
||||||
|
HasReadSeq: seq,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {}
|
||||||
|
}
|
||||||
@@ -24,6 +24,7 @@ import (
|
|||||||
"github.com/openimsdk/protocol/constant"
|
"github.com/openimsdk/protocol/constant"
|
||||||
"github.com/openimsdk/protocol/sdkws"
|
"github.com/openimsdk/protocol/sdkws"
|
||||||
"github.com/openimsdk/tools/mcontext"
|
"github.com/openimsdk/tools/mcontext"
|
||||||
|
"github.com/openimsdk/tools/utils/datautil"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (c *ConsumerHandler) webhookBeforeOfflinePush(ctx context.Context, before *config.BeforeConfig, userIDs []string, msg *sdkws.MsgData, offlinePushUserIDs *[]string) error {
|
func (c *ConsumerHandler) webhookBeforeOfflinePush(ctx context.Context, before *config.BeforeConfig, userIDs []string, msg *sdkws.MsgData, offlinePushUserIDs *[]string) error {
|
||||||
@@ -69,7 +70,7 @@ func (c *ConsumerHandler) webhookBeforeOfflinePush(ctx context.Context, before *
|
|||||||
|
|
||||||
func (c *ConsumerHandler) webhookBeforeOnlinePush(ctx context.Context, before *config.BeforeConfig, userIDs []string, msg *sdkws.MsgData) error {
|
func (c *ConsumerHandler) webhookBeforeOnlinePush(ctx context.Context, before *config.BeforeConfig, userIDs []string, msg *sdkws.MsgData) error {
|
||||||
return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
|
return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
|
||||||
if msg.ContentType == constant.Typing {
|
if datautil.Contain(msg.SendID, userIDs...) || msg.ContentType == constant.Typing {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
req := callbackstruct.CallbackBeforePushReq{
|
req := callbackstruct.CallbackBeforePushReq{
|
||||||
|
|||||||
@@ -23,13 +23,10 @@ import (
|
|||||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
"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/config"
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||||
"github.com/openimsdk/tools/log"
|
|
||||||
"github.com/openimsdk/tools/mcontext"
|
|
||||||
"strings"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
geTUI = "getui"
|
geTUI = "geTui"
|
||||||
firebase = "fcm"
|
firebase = "fcm"
|
||||||
jPush = "jpush"
|
jPush = "jpush"
|
||||||
)
|
)
|
||||||
@@ -41,7 +38,6 @@ type OfflinePusher interface {
|
|||||||
|
|
||||||
func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (OfflinePusher, error) {
|
func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPath string) (OfflinePusher, error) {
|
||||||
var offlinePusher OfflinePusher
|
var offlinePusher OfflinePusher
|
||||||
pushConf.Enable = strings.ToLower(pushConf.Enable)
|
|
||||||
switch pushConf.Enable {
|
switch pushConf.Enable {
|
||||||
case geTUI:
|
case geTUI:
|
||||||
offlinePusher = getui.NewClient(pushConf, cache)
|
offlinePusher = getui.NewClient(pushConf, cache)
|
||||||
@@ -51,7 +47,6 @@ func NewOfflinePusher(pushConf *config.Push, cache cache.ThirdCache, fcmConfigPa
|
|||||||
offlinePusher = jpush.NewClient(pushConf)
|
offlinePusher = jpush.NewClient(pushConf)
|
||||||
default:
|
default:
|
||||||
offlinePusher = dummy.NewClient()
|
offlinePusher = dummy.NewClient()
|
||||||
log.ZWarn(mcontext.WithMustInfoCtx([]string{"push start", "admin", "admin", ""}), "Unknown push config", nil)
|
|
||||||
}
|
}
|
||||||
return offlinePusher, nil
|
return offlinePusher, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -55,9 +55,6 @@ func (o *OfflinePushConsumerHandler) handleMsg2OfflinePush(ctx context.Context,
|
|||||||
log.ZError(ctx, "offline push msg is empty", errs.New("offlinePushMsg is empty"), "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
|
log.ZError(ctx, "offline push msg is empty", errs.New("offlinePushMsg is empty"), "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if offlinePushMsg.MsgData.Status == constant.MsgStatusSending {
|
|
||||||
offlinePushMsg.MsgData.Status = constant.MsgStatusSendSuccess
|
|
||||||
}
|
|
||||||
log.ZInfo(ctx, "receive to OfflinePush MQ", "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
|
log.ZInfo(ctx, "receive to OfflinePush MQ", "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
|
||||||
|
|
||||||
err := o.offlinePushMsg(ctx, offlinePushMsg.MsgData, offlinePushMsg.UserIDs)
|
err := o.offlinePushMsg(ctx, offlinePushMsg.MsgData, offlinePushMsg.UserIDs)
|
||||||
|
|||||||
@@ -194,9 +194,6 @@ func (c *ConsumerHandler) shouldPushOffline(_ context.Context, msg *sdkws.MsgDat
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) ([]*msggateway.SingleMsgToUserResults, error) {
|
func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) ([]*msggateway.SingleMsgToUserResults, error) {
|
||||||
if msg != nil && msg.Status == constant.MsgStatusSending {
|
|
||||||
msg.Status = constant.MsgStatusSendSuccess
|
|
||||||
}
|
|
||||||
onlineUserIDs, offlineUserIDs, err := c.onlineCache.GetUsersOnline(ctx, pushToUserIDs)
|
onlineUserIDs, offlineUserIDs, err := c.onlineCache.GetUsersOnline(ctx, pushToUserIDs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|||||||
@@ -65,7 +65,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
|
|||||||
redis2.NewTokenCacheModel(rdb, config.RpcConfig.TokenPolicy.Expire),
|
redis2.NewTokenCacheModel(rdb, config.RpcConfig.TokenPolicy.Expire),
|
||||||
config.Share.Secret,
|
config.Share.Secret,
|
||||||
config.RpcConfig.TokenPolicy.Expire,
|
config.RpcConfig.TokenPolicy.Expire,
|
||||||
config.Share.MultiLogin,
|
config.Share.MultiLoginPolicy,
|
||||||
),
|
),
|
||||||
config: config,
|
config: config,
|
||||||
})
|
})
|
||||||
@@ -230,10 +230,3 @@ func (s *authServer) InvalidateToken(ctx context.Context, req *pbauth.Invalidate
|
|||||||
}
|
}
|
||||||
return &pbauth.InvalidateTokenResp{}, nil
|
return &pbauth.InvalidateTokenResp{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *authServer) KickTokens(ctx context.Context, req *pbauth.KickTokensReq) (*pbauth.KickTokensResp, error) {
|
|
||||||
if err := s.authDatabase.BatchSetTokenMapByUidPid(ctx, req.Tokens); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return &pbauth.KickTokensResp{}, nil
|
|
||||||
}
|
|
||||||
|
|||||||
+29
-83
@@ -465,7 +465,7 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, opUserID, req.InvitedUserIDs...); err != nil {
|
if err = g.notification.MemberEnterNotification(ctx, req.GroupID, req.InvitedUserIDs...); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return &pbgroup.InviteUserToGroupResp{}, nil
|
return &pbgroup.InviteUserToGroupResp{}, nil
|
||||||
@@ -1026,7 +1026,7 @@ func (g *groupServer) SetGroupInfo(ctx context.Context, req *pbgroup.SetGroupInf
|
|||||||
}
|
}
|
||||||
num := len(update)
|
num := len(update)
|
||||||
if req.GroupInfoForSet.Notification != "" {
|
if req.GroupInfoForSet.Notification != "" {
|
||||||
num -= 3
|
num--
|
||||||
func() {
|
func() {
|
||||||
conversation := &pbconversation.ConversationReq{
|
conversation := &pbconversation.ConversationReq{
|
||||||
ConversationID: msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupInfoForSet.GroupID),
|
ConversationID: msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupInfoForSet.GroupID),
|
||||||
@@ -1133,9 +1133,8 @@ func (g *groupServer) SetGroupInfoEx(ctx context.Context, req *pbgroup.SetGroupI
|
|||||||
}
|
}
|
||||||
|
|
||||||
num := len(updatedData)
|
num := len(updatedData)
|
||||||
|
|
||||||
if req.Notification != nil {
|
if req.Notification != nil {
|
||||||
num -= 3
|
num--
|
||||||
|
|
||||||
if req.Notification.Value != "" {
|
if req.Notification.Value != "" {
|
||||||
func() {
|
func() {
|
||||||
@@ -1181,53 +1180,36 @@ func (g *groupServer) TransferGroupOwner(ctx context.Context, req *pbgroup.Trans
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if group.Status == constant.GroupStatusDismissed {
|
if group.Status == constant.GroupStatusDismissed {
|
||||||
return nil, servererrs.ErrDismissedAlready.Wrap()
|
return nil, servererrs.ErrDismissedAlready.Wrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
if req.OldOwnerUserID == req.NewOwnerUserID {
|
if req.OldOwnerUserID == req.NewOwnerUserID {
|
||||||
return nil, errs.ErrArgs.WrapMsg("OldOwnerUserID == NewOwnerUserID")
|
return nil, errs.ErrArgs.WrapMsg("OldOwnerUserID == NewOwnerUserID")
|
||||||
}
|
}
|
||||||
|
|
||||||
members, err := g.db.FindGroupMembers(ctx, req.GroupID, []string{req.OldOwnerUserID, req.NewOwnerUserID})
|
members, err := g.db.FindGroupMembers(ctx, req.GroupID, []string{req.OldOwnerUserID, req.NewOwnerUserID})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := g.PopulateGroupMember(ctx, members...); err != nil {
|
if err := g.PopulateGroupMember(ctx, members...); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
memberMap := datautil.SliceToMap(members, func(e *model.GroupMember) string { return e.UserID })
|
memberMap := datautil.SliceToMap(members, func(e *model.GroupMember) string { return e.UserID })
|
||||||
if ids := datautil.Single([]string{req.OldOwnerUserID, req.NewOwnerUserID}, datautil.Keys(memberMap)); len(ids) > 0 {
|
if ids := datautil.Single([]string{req.OldOwnerUserID, req.NewOwnerUserID}, datautil.Keys(memberMap)); len(ids) > 0 {
|
||||||
return nil, errs.ErrArgs.WrapMsg("user not in group " + strings.Join(ids, ","))
|
return nil, errs.ErrArgs.WrapMsg("user not in group " + strings.Join(ids, ","))
|
||||||
}
|
}
|
||||||
|
|
||||||
oldOwner := memberMap[req.OldOwnerUserID]
|
oldOwner := memberMap[req.OldOwnerUserID]
|
||||||
if oldOwner == nil {
|
if oldOwner == nil {
|
||||||
return nil, errs.ErrArgs.WrapMsg("OldOwnerUserID not in group " + req.NewOwnerUserID)
|
return nil, errs.ErrArgs.WrapMsg("OldOwnerUserID not in group " + req.NewOwnerUserID)
|
||||||
}
|
}
|
||||||
|
|
||||||
newOwner := memberMap[req.NewOwnerUserID]
|
newOwner := memberMap[req.NewOwnerUserID]
|
||||||
if newOwner == nil {
|
if newOwner == nil {
|
||||||
return nil, errs.ErrArgs.WrapMsg("NewOwnerUser not in group " + req.NewOwnerUserID)
|
return nil, errs.ErrArgs.WrapMsg("NewOwnerUser not in group " + req.NewOwnerUserID)
|
||||||
}
|
}
|
||||||
|
|
||||||
if !authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID) {
|
if !authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID) {
|
||||||
if !(mcontext.GetOpUserID(ctx) == oldOwner.UserID && oldOwner.RoleLevel == constant.GroupOwner) {
|
if !(mcontext.GetOpUserID(ctx) == oldOwner.UserID && oldOwner.RoleLevel == constant.GroupOwner) {
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("no permission transfer group owner")
|
return nil, errs.ErrNoPermission.WrapMsg("no permission transfer group owner")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if newOwner.MuteEndTime.After(time.Now()) {
|
|
||||||
if _, err := g.CancelMuteGroupMember(ctx, &pbgroup.CancelMuteGroupMemberReq{
|
|
||||||
GroupID: group.GroupID,
|
|
||||||
UserID: req.NewOwnerUserID}); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := g.db.TransferGroupOwner(ctx, req.GroupID, req.OldOwnerUserID, req.NewOwnerUserID, newOwner.RoleLevel); err != nil {
|
if err := g.db.TransferGroupOwner(ctx, req.GroupID, req.OldOwnerUserID, req.NewOwnerUserID, newOwner.RoleLevel); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -1235,7 +1217,6 @@ func (g *groupServer) TransferGroupOwner(ctx context.Context, req *pbgroup.Trans
|
|||||||
g.webhookAfterTransferGroupOwner(ctx, &g.config.WebhooksConfig.AfterTransferGroupOwner, req)
|
g.webhookAfterTransferGroupOwner(ctx, &g.config.WebhooksConfig.AfterTransferGroupOwner, req)
|
||||||
|
|
||||||
g.notification.GroupOwnerTransferredNotification(ctx, req)
|
g.notification.GroupOwnerTransferredNotification(ctx, req)
|
||||||
|
|
||||||
return &pbgroup.TransferGroupOwnerResp{}, nil
|
return &pbgroup.TransferGroupOwnerResp{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1444,38 +1425,32 @@ func (g *groupServer) CancelMuteGroupMember(ctx context.Context, req *pbgroup.Ca
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := g.PopulateGroupMember(ctx, member); err != nil {
|
if err := g.PopulateGroupMember(ctx, member); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if !authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID) {
|
if !authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID) {
|
||||||
opMember, err := g.db.TakeGroupMember(ctx, req.GroupID, mcontext.GetOpUserID(ctx))
|
opMember, err := g.db.TakeGroupMember(ctx, req.GroupID, mcontext.GetOpUserID(ctx))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
switch member.RoleLevel {
|
switch member.RoleLevel {
|
||||||
case constant.GroupOwner:
|
case constant.GroupOwner:
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("Can not set group owner unmute")
|
return nil, errs.ErrNoPermission.WrapMsg("set group owner mute")
|
||||||
case constant.GroupAdmin:
|
case constant.GroupAdmin:
|
||||||
if opMember.RoleLevel != constant.GroupOwner {
|
if opMember.RoleLevel != constant.GroupOwner {
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("Can not set group admin unmute")
|
return nil, errs.ErrNoPermission.WrapMsg("set group admin mute")
|
||||||
}
|
}
|
||||||
case constant.GroupOrdinaryUsers:
|
case constant.GroupOrdinaryUsers:
|
||||||
if !(opMember.RoleLevel == constant.GroupAdmin || opMember.RoleLevel == constant.GroupOwner) {
|
if !(opMember.RoleLevel == constant.GroupAdmin || opMember.RoleLevel == constant.GroupOwner) {
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("Can not set group ordinary users unmute")
|
return nil, errs.ErrNoPermission.WrapMsg("set group ordinary users mute")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
data := UpdateGroupMemberMutedTimeMap(time.Unix(0, 0))
|
data := UpdateGroupMemberMutedTimeMap(time.Unix(0, 0))
|
||||||
if err := g.db.UpdateGroupMember(ctx, member.GroupID, member.UserID, data); err != nil {
|
if err := g.db.UpdateGroupMember(ctx, member.GroupID, member.UserID, data); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
g.notification.GroupMemberCancelMutedNotification(ctx, req.GroupID, req.UserID)
|
g.notification.GroupMemberCancelMutedNotification(ctx, req.GroupID, req.UserID)
|
||||||
|
|
||||||
return &pbgroup.CancelMuteGroupMemberResp{}, nil
|
return &pbgroup.CancelMuteGroupMemberResp{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1510,6 +1485,9 @@ func (g *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbgroup.SetGr
|
|||||||
return nil, errs.ErrNoPermission.WrapMsg("no op user id")
|
return nil, errs.ErrNoPermission.WrapMsg("no op user id")
|
||||||
}
|
}
|
||||||
isAppManagerUid := authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID)
|
isAppManagerUid := authverify.IsAppManagerUid(ctx, g.config.Share.IMAdminUserID)
|
||||||
|
for i := range req.Members {
|
||||||
|
req.Members[i].FaceURL = nil
|
||||||
|
}
|
||||||
groupMembers := make(map[string][]*pbgroup.SetGroupMemberInfo)
|
groupMembers := make(map[string][]*pbgroup.SetGroupMemberInfo)
|
||||||
for i, member := range req.Members {
|
for i, member := range req.Members {
|
||||||
if member.RoleLevel != nil {
|
if member.RoleLevel != nil {
|
||||||
@@ -1551,61 +1529,29 @@ func (g *groupServer) SetGroupMemberInfo(ctx context.Context, req *pbgroup.SetGr
|
|||||||
case 0:
|
case 0:
|
||||||
if !isAppManagerUid {
|
if !isAppManagerUid {
|
||||||
roleLevel := dbMembers[opUserIndex].RoleLevel
|
roleLevel := dbMembers[opUserIndex].RoleLevel
|
||||||
var (
|
if roleLevel != constant.GroupOwner {
|
||||||
dbSelf = &model.GroupMember{}
|
switch roleLevel {
|
||||||
reqSelf *pbgroup.SetGroupMemberInfo
|
case constant.GroupAdmin:
|
||||||
)
|
for _, member := range dbMembers {
|
||||||
switch roleLevel {
|
if member.RoleLevel == constant.GroupOwner {
|
||||||
case constant.GroupOwner:
|
return nil, errs.ErrNoPermission.WrapMsg("admin can not change group owner")
|
||||||
for _, member := range dbMembers {
|
}
|
||||||
if member.UserID == opUserID {
|
if member.RoleLevel == constant.GroupAdmin && member.UserID != opUserID {
|
||||||
dbSelf = member
|
return nil, errs.ErrNoPermission.WrapMsg("admin can not change other group admin")
|
||||||
break
|
}
|
||||||
}
|
}
|
||||||
}
|
case constant.GroupOrdinaryUsers:
|
||||||
case constant.GroupAdmin:
|
for _, member := range dbMembers {
|
||||||
for _, member := range dbMembers {
|
if !(member.RoleLevel == constant.GroupOrdinaryUsers && member.UserID == opUserID) {
|
||||||
if member.UserID == opUserID {
|
return nil, errs.ErrNoPermission.WrapMsg("ordinary users can not change other role level")
|
||||||
dbSelf = member
|
}
|
||||||
}
|
}
|
||||||
if member.RoleLevel == constant.GroupOwner {
|
default:
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("admin can not change group owner")
|
for _, member := range dbMembers {
|
||||||
|
if member.RoleLevel >= roleLevel {
|
||||||
|
return nil, errs.ErrNoPermission.WrapMsg("can not change higher role level")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if member.RoleLevel == constant.GroupAdmin && member.UserID != opUserID {
|
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("admin can not change other group admin")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case constant.GroupOrdinaryUsers:
|
|
||||||
for _, member := range dbMembers {
|
|
||||||
if member.UserID == opUserID {
|
|
||||||
dbSelf = member
|
|
||||||
}
|
|
||||||
if !(member.RoleLevel == constant.GroupOrdinaryUsers && member.UserID == opUserID) {
|
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("ordinary users can not change other role level")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
default:
|
|
||||||
for _, member := range dbMembers {
|
|
||||||
if member.UserID == opUserID {
|
|
||||||
dbSelf = member
|
|
||||||
}
|
|
||||||
if member.RoleLevel >= roleLevel {
|
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("can not change higher role level")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, member := range req.Members {
|
|
||||||
if member.UserID == opUserID {
|
|
||||||
reqSelf = member
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if reqSelf != nil && reqSelf.RoleLevel != nil {
|
|
||||||
if reqSelf.RoleLevel.GetValue() > dbSelf.RoleLevel {
|
|
||||||
return nil, errs.ErrNoPermission.WrapMsg("can not improve role level by self")
|
|
||||||
}
|
|
||||||
if roleLevel == constant.GroupOwner {
|
|
||||||
return nil, errs.ErrArgs.WrapMsg("group owner can not change own role level") // Prevent the absence of a group owner
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -38,7 +38,6 @@ import (
|
|||||||
"github.com/openimsdk/tools/utils/datautil"
|
"github.com/openimsdk/tools/utils/datautil"
|
||||||
"github.com/openimsdk/tools/utils/stringutil"
|
"github.com/openimsdk/tools/utils/stringutil"
|
||||||
"go.mongodb.org/mongo-driver/mongo"
|
"go.mongodb.org/mongo-driver/mongo"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// GroupApplicationReceiver
|
// GroupApplicationReceiver
|
||||||
@@ -573,51 +572,8 @@ func (g *GroupNotificationSender) GroupApplicationAgreeMemberEnterNotification(c
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, groupID string, entrantUserID string) error {
|
func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, groupID string, entrantUserID ...string) error {
|
||||||
var err error
|
return g.GroupApplicationAgreeMemberEnterNotification(ctx, groupID, "", entrantUserID...)
|
||||||
defer func() {
|
|
||||||
if err != nil {
|
|
||||||
log.ZError(ctx, stringutil.GetFuncName(1)+" failed", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
if !g.config.RpcConfig.EnableHistoryForNewMembers {
|
|
||||||
conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID)
|
|
||||||
maxSeq, err := g.msgRpcClient.GetConversationMaxSeq(ctx, conversationID)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if _, err = g.msgRpcClient.SetUserConversationsMinSeq(ctx, &msg.SetUserConversationsMinSeqReq{
|
|
||||||
UserIDs: []string{entrantUserID},
|
|
||||||
ConversationID: conversationID,
|
|
||||||
Seq: maxSeq,
|
|
||||||
}); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := g.conversationRpcClient.GroupChatFirstCreateConversation(ctx, groupID, []string{entrantUserID}); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
var group *sdkws.GroupInfo
|
|
||||||
group, err = g.getGroupInfo(ctx, groupID)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
user, err := g.getGroupMember(ctx, groupID, entrantUserID)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
tips := &sdkws.MemberEnterTips{
|
|
||||||
Group: group,
|
|
||||||
EntrantUser: user,
|
|
||||||
OperationTime: time.Now().UnixMilli(),
|
|
||||||
}
|
|
||||||
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
|
|
||||||
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberEnterNotification, tips)
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (g *GroupNotificationSender) GroupDismissedNotification(ctx context.Context, tips *sdkws.GroupDismissedTips) {
|
func (g *GroupNotificationSender) GroupDismissedNotification(ctx context.Context, tips *sdkws.GroupDismissedTips) {
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ import (
|
|||||||
"github.com/openimsdk/tools/log"
|
"github.com/openimsdk/tools/log"
|
||||||
"github.com/openimsdk/tools/mcontext"
|
"github.com/openimsdk/tools/mcontext"
|
||||||
"github.com/openimsdk/tools/utils/datautil"
|
"github.com/openimsdk/tools/utils/datautil"
|
||||||
|
"github.com/openimsdk/tools/utils/stringutil"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
|
func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
|
||||||
@@ -79,17 +80,13 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *pbmsg.SendMsgReq)
|
|||||||
|
|
||||||
func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgData) {
|
func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgData) {
|
||||||
log.ZDebug(nctx, "setConversationAtInfo", "msg", msg)
|
log.ZDebug(nctx, "setConversationAtInfo", "msg", msg)
|
||||||
|
|
||||||
ctx := mcontext.NewCtx("@@@" + mcontext.GetOperationID(nctx))
|
ctx := mcontext.NewCtx("@@@" + mcontext.GetOperationID(nctx))
|
||||||
|
|
||||||
var atUserID []string
|
var atUserID []string
|
||||||
|
|
||||||
conversation := &pbconversation.ConversationReq{
|
conversation := &pbconversation.ConversationReq{
|
||||||
ConversationID: msgprocessor.GetConversationIDByMsg(msg),
|
ConversationID: msgprocessor.GetConversationIDByMsg(msg),
|
||||||
ConversationType: msg.SessionType,
|
ConversationType: msg.SessionType,
|
||||||
GroupID: msg.GroupID,
|
GroupID: msg.GroupID,
|
||||||
}
|
}
|
||||||
|
|
||||||
tagAll := datautil.Contain(constant.AtAllString, msg.AtUserIDList...)
|
tagAll := datautil.Contain(constant.AtAllString, msg.AtUserIDList...)
|
||||||
if tagAll {
|
if tagAll {
|
||||||
memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
|
memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
|
||||||
@@ -97,35 +94,25 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
|
|||||||
log.ZWarn(ctx, "GetGroupMemberIDs", err)
|
log.ZWarn(ctx, "GetGroupMemberIDs", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
atUserID = stringutil.DifferenceString([]string{constant.AtAllString}, msg.AtUserIDList)
|
||||||
memberUserIDList = datautil.DeleteElems(memberUserIDList, msg.SendID)
|
|
||||||
|
|
||||||
atUserID = datautil.Single([]string{constant.AtAllString}, msg.AtUserIDList)
|
|
||||||
|
|
||||||
if len(atUserID) == 0 { // just @everyone
|
if len(atUserID) == 0 { // just @everyone
|
||||||
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
|
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
|
||||||
} else { // @Everyone and @other people
|
} else { // @Everyone and @other people
|
||||||
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAllAtMe}
|
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAllAtMe}
|
||||||
|
|
||||||
err = m.Conversation.SetConversations(ctx, atUserID, conversation)
|
err = m.Conversation.SetConversations(ctx, atUserID, conversation)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation)
|
log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation)
|
||||||
}
|
}
|
||||||
|
memberUserIDList = stringutil.DifferenceString(atUserID, memberUserIDList)
|
||||||
memberUserIDList = datautil.Single(atUserID, memberUserIDList)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
|
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
|
||||||
|
|
||||||
err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation)
|
err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.ZWarn(ctx, "SetConversations", err, "userID", memberUserIDList, "conversation", conversation)
|
log.ZWarn(ctx, "SetConversations", err, "userID", memberUserIDList, "conversation", conversation)
|
||||||
}
|
}
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtMe}
|
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtMe}
|
||||||
|
|
||||||
err := m.Conversation.SetConversations(ctx, msg.AtUserIDList, conversation)
|
err := m.Conversation.SetConversations(ctx, msg.AtUserIDList, conversation)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.ZWarn(ctx, "SetConversations", err, msg.AtUserIDList, conversation)
|
log.ZWarn(ctx, "SetConversations", err, msg.AtUserIDList, conversation)
|
||||||
|
|||||||
@@ -116,17 +116,18 @@ func Start(ctx context.Context, config *Config, client registry.SvcDiscoveryRegi
|
|||||||
|
|
||||||
func (s *userServer) GetDesignateUsers(ctx context.Context, req *pbuser.GetDesignateUsersReq) (resp *pbuser.GetDesignateUsersResp, err error) {
|
func (s *userServer) GetDesignateUsers(ctx context.Context, req *pbuser.GetDesignateUsersReq) (resp *pbuser.GetDesignateUsersResp, err error) {
|
||||||
resp = &pbuser.GetDesignateUsersResp{}
|
resp = &pbuser.GetDesignateUsersResp{}
|
||||||
users, err := s.db.Find(ctx, req.UserIDs)
|
users, err := s.db.FindWithError(ctx, req.UserIDs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
resp.UsersInfo = convert.UsersDB2Pb(users)
|
resp.UsersInfo = convert.UsersDB2Pb(users)
|
||||||
return resp, nil
|
return resp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// deprecated:
|
// deprecated:
|
||||||
// UpdateUserInfo
|
|
||||||
|
//UpdateUserInfo
|
||||||
|
|
||||||
func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfoReq) (resp *pbuser.UpdateUserInfoResp, err error) {
|
func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserInfoReq) (resp *pbuser.UpdateUserInfoResp, err error) {
|
||||||
resp = &pbuser.UpdateUserInfoResp{}
|
resp = &pbuser.UpdateUserInfoResp{}
|
||||||
err = authverify.CheckAccessV3(ctx, req.UserInfo.UserID, s.config.Share.IMAdminUserID)
|
err = authverify.CheckAccessV3(ctx, req.UserInfo.UserID, s.config.Share.IMAdminUserID)
|
||||||
|
|||||||
@@ -361,29 +361,11 @@ type AfterConfig struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type Share struct {
|
type Share struct {
|
||||||
Secret string `mapstructure:"secret"`
|
Secret string `mapstructure:"secret"`
|
||||||
RpcRegisterName RpcRegisterName `mapstructure:"rpcRegisterName"`
|
RpcRegisterName RpcRegisterName `mapstructure:"rpcRegisterName"`
|
||||||
IMAdminUserID []string `mapstructure:"imAdminUserID"`
|
IMAdminUserID []string `mapstructure:"imAdminUserID"`
|
||||||
MultiLogin MultiLogin `mapstructure:"multiLogin"`
|
MultiLoginPolicy int `mapstructure:"multiLoginPolicy"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type MultiLogin struct {
|
|
||||||
Policy int `mapstructure:"policy"`
|
|
||||||
MaxNumOneEnd int `mapstructure:"maxNumOneEnd"`
|
|
||||||
CustomizeLoginNum struct {
|
|
||||||
IOS int `mapstructure:"ios"`
|
|
||||||
Android int `mapstructure:"android"`
|
|
||||||
Windows int `mapstructure:"windows"`
|
|
||||||
OSX int `mapstructure:"osx"`
|
|
||||||
Web int `mapstructure:"web"`
|
|
||||||
MiniWeb int `mapstructure:"miniWeb"`
|
|
||||||
Linux int `mapstructure:"linux"`
|
|
||||||
APad int `mapstructure:"aPad"`
|
|
||||||
IPad int `mapstructure:"iPad"`
|
|
||||||
Admin int `mapstructure:"admin"`
|
|
||||||
} `mapstructure:"customizeLoginNum"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type RpcRegisterName struct {
|
type RpcRegisterName struct {
|
||||||
User string `mapstructure:"user"`
|
User string `mapstructure:"user"`
|
||||||
Friend string `mapstructure:"friend"`
|
Friend string `mapstructure:"friend"`
|
||||||
|
|||||||
+1
-18
@@ -1,9 +1,6 @@
|
|||||||
package cachekey
|
package cachekey
|
||||||
|
|
||||||
import (
|
import "github.com/openimsdk/protocol/constant"
|
||||||
"github.com/openimsdk/protocol/constant"
|
|
||||||
"strings"
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
const (
|
||||||
UidPidToken = "UID_PID_TOKEN_STATUS:"
|
UidPidToken = "UID_PID_TOKEN_STATUS:"
|
||||||
@@ -12,17 +9,3 @@ const (
|
|||||||
func GetTokenKey(userID string, platformID int) string {
|
func GetTokenKey(userID string, platformID int) string {
|
||||||
return UidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
|
return UidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
|
||||||
}
|
}
|
||||||
|
|
||||||
func GetAllPlatformTokenKey(userID string) []string {
|
|
||||||
res := make([]string, len(constant.PlatformID2Name))
|
|
||||||
for k := range constant.PlatformID2Name {
|
|
||||||
res[k-1] = GetTokenKey(userID, k)
|
|
||||||
}
|
|
||||||
return res
|
|
||||||
}
|
|
||||||
|
|
||||||
func GetPlatformIDByTokenKey(key string) int {
|
|
||||||
splitKey := strings.Split(key, ":")
|
|
||||||
platform := splitKey[len(splitKey)-1]
|
|
||||||
return constant.PlatformNameToID(platform)
|
|
||||||
}
|
|
||||||
|
|||||||
+14
-50
@@ -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
|
package redis
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -7,7 +21,6 @@ import (
|
|||||||
"github.com/openimsdk/tools/errs"
|
"github.com/openimsdk/tools/errs"
|
||||||
"github.com/redis/go-redis/v9"
|
"github.com/redis/go-redis/v9"
|
||||||
"strconv"
|
"strconv"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -54,43 +67,6 @@ func (c *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
|
|||||||
return mm, nil
|
return mm, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
|
|
||||||
var (
|
|
||||||
res = make(map[int]map[string]int)
|
|
||||||
resLock = sync.Mutex{}
|
|
||||||
)
|
|
||||||
|
|
||||||
keys := cachekey.GetAllPlatformTokenKey(userID)
|
|
||||||
if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
|
|
||||||
pipe := c.rdb.Pipeline()
|
|
||||||
mapRes := make([]*redis.MapStringStringCmd, len(keys))
|
|
||||||
for i, key := range keys {
|
|
||||||
mapRes[i] = pipe.HGetAll(ctx, key)
|
|
||||||
}
|
|
||||||
_, err := pipe.Exec(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
for i, m := range mapRes {
|
|
||||||
mm := make(map[string]int)
|
|
||||||
for k, v := range m.Val() {
|
|
||||||
state, err := strconv.Atoi(v)
|
|
||||||
if err != nil {
|
|
||||||
return errs.WrapMsg(err, "redis token value is not int", "value", v, "userID", userID)
|
|
||||||
}
|
|
||||||
mm[k] = state
|
|
||||||
}
|
|
||||||
resLock.Lock()
|
|
||||||
res[cachekey.GetPlatformIDByTokenKey(keys[i])] = mm
|
|
||||||
resLock.Unlock()
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return res, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
|
func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
|
||||||
mm := make(map[string]any)
|
mm := make(map[string]any)
|
||||||
for k, v := range m {
|
for k, v := range m {
|
||||||
@@ -99,18 +75,6 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla
|
|||||||
return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err())
|
return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err())
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]int) error {
|
|
||||||
pipe := c.rdb.Pipeline()
|
|
||||||
for k, v := range tokens {
|
|
||||||
pipe.HSet(ctx, k, v)
|
|
||||||
}
|
|
||||||
_, err := pipe.Exec(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return errs.Wrap(err)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error {
|
func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error {
|
||||||
return errs.Wrap(c.rdb.HDel(ctx, cachekey.GetTokenKey(userID, platformID), fields...).Err())
|
return errs.Wrap(c.rdb.HDel(ctx, cachekey.GetTokenKey(userID, platformID), fields...).Err())
|
||||||
}
|
}
|
||||||
|
|||||||
Vendored
-2
@@ -9,8 +9,6 @@ type TokenModel interface {
|
|||||||
// SetTokenFlagEx set token and flag with expire time
|
// SetTokenFlagEx set token and flag with expire time
|
||||||
SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error
|
SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error
|
||||||
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
|
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
|
||||||
GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error)
|
|
||||||
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
|
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
|
||||||
BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]int) error
|
|
||||||
DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
|
DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,9 +1,21 @@
|
|||||||
|
// 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 controller
|
package controller
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"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/openimsdk/tools/log"
|
||||||
|
|
||||||
"github.com/golang-jwt/jwt/v4"
|
"github.com/golang-jwt/jwt/v4"
|
||||||
@@ -20,41 +32,18 @@ type AuthDatabase interface {
|
|||||||
// Create token
|
// Create token
|
||||||
CreateToken(ctx context.Context, userID string, platformID int) (string, error)
|
CreateToken(ctx context.Context, userID string, platformID int) (string, error)
|
||||||
|
|
||||||
BatchSetTokenMapByUidPid(ctx context.Context, tokens []string) error
|
|
||||||
|
|
||||||
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
|
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
|
||||||
}
|
}
|
||||||
|
|
||||||
type multiLoginConfig struct {
|
|
||||||
Policy int
|
|
||||||
MaxNumOneEnd int
|
|
||||||
CustomizeLoginNum map[int]int
|
|
||||||
}
|
|
||||||
|
|
||||||
type authDatabase struct {
|
type authDatabase struct {
|
||||||
cache cache.TokenModel
|
cache cache.TokenModel
|
||||||
accessSecret string
|
accessSecret string
|
||||||
accessExpire int64
|
accessExpire int64
|
||||||
multiLogin multiLoginConfig
|
multiLoginPolicy int
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, multiLogin config.MultiLogin) AuthDatabase {
|
func NewAuthDatabase(cache cache.TokenModel, accessSecret string, accessExpire int64, policy int) AuthDatabase {
|
||||||
return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLogin: multiLoginConfig{
|
return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire, multiLoginPolicy: policy}
|
||||||
Policy: multiLogin.Policy,
|
|
||||||
MaxNumOneEnd: multiLogin.MaxNumOneEnd,
|
|
||||||
CustomizeLoginNum: map[int]int{
|
|
||||||
constant.IOSPlatformID: multiLogin.CustomizeLoginNum.IOS,
|
|
||||||
constant.AndroidPlatformID: multiLogin.CustomizeLoginNum.Android,
|
|
||||||
constant.WindowsPlatformID: multiLogin.CustomizeLoginNum.Windows,
|
|
||||||
constant.OSXPlatformID: multiLogin.CustomizeLoginNum.OSX,
|
|
||||||
constant.WebPlatformID: multiLogin.CustomizeLoginNum.Web,
|
|
||||||
constant.MiniWebPlatformID: multiLogin.CustomizeLoginNum.MiniWeb,
|
|
||||||
constant.LinuxPlatformID: multiLogin.CustomizeLoginNum.Linux,
|
|
||||||
constant.AndroidPadPlatformID: multiLogin.CustomizeLoginNum.APad,
|
|
||||||
constant.IPadPlatformID: multiLogin.CustomizeLoginNum.IPad,
|
|
||||||
constant.AdminPlatformID: multiLogin.CustomizeLoginNum.Admin,
|
|
||||||
},
|
|
||||||
}}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// If the result is empty.
|
// If the result is empty.
|
||||||
@@ -66,38 +55,22 @@ func (a *authDatabase) SetTokenMapByUidPid(ctx context.Context, userID string, p
|
|||||||
return a.cache.SetTokenMapByUidPid(ctx, userID, platformID, m)
|
return a.cache.SetTokenMapByUidPid(ctx, userID, platformID, m)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *authDatabase) BatchSetTokenMapByUidPid(ctx context.Context, tokens []string) error {
|
|
||||||
setMap := make(map[string]map[string]int)
|
|
||||||
for _, token := range tokens {
|
|
||||||
claims, err := tokenverify.GetClaimFromToken(token, authverify.Secret(a.accessSecret))
|
|
||||||
key := cachekey.GetTokenKey(claims.UserID, claims.PlatformID)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
} else {
|
|
||||||
if v, ok := setMap[key]; ok {
|
|
||||||
v[token] = constant.KickedToken
|
|
||||||
} else {
|
|
||||||
setMap[key] = map[string]int{
|
|
||||||
token: constant.KickedToken,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err := a.cache.BatchSetTokenMapByUidPid(ctx, setMap); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Create Token.
|
// Create Token.
|
||||||
func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) {
|
func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) {
|
||||||
tokens, err := a.cache.GetAllTokensWithoutError(ctx, userID)
|
// todo: get all platform token
|
||||||
|
tokens, err := a.cache.GetTokensWithoutError(ctx, userID, platformID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
|
var deleteTokenKey []string
|
||||||
if err != nil {
|
var kickedTokenKey []string
|
||||||
return "", err
|
for k, v := range tokens {
|
||||||
|
t, err := tokenverify.GetClaimFromToken(k, authverify.Secret(a.accessSecret))
|
||||||
|
if err != nil || v != constant.NormalToken {
|
||||||
|
deleteTokenKey = append(deleteTokenKey, k)
|
||||||
|
} else if a.checkKickToken(ctx, platformID, t) {
|
||||||
|
kickedTokenKey = append(kickedTokenKey, k)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if len(deleteTokenKey) != 0 {
|
if len(deleteTokenKey) != 0 {
|
||||||
err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
|
err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
|
||||||
@@ -105,6 +78,16 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
|
|||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const adminTokenMaxNum = 30
|
||||||
|
if platformID == constant.AdminPlatformID {
|
||||||
|
if len(kickedTokenKey) > adminTokenMaxNum {
|
||||||
|
kickedTokenKey = kickedTokenKey[:len(kickedTokenKey)-adminTokenMaxNum]
|
||||||
|
} else {
|
||||||
|
kickedTokenKey = nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if len(kickedTokenKey) != 0 {
|
if len(kickedTokenKey) != 0 {
|
||||||
for _, k := range kickedTokenKey {
|
for _, k := range kickedTokenKey {
|
||||||
err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
|
err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
|
||||||
@@ -128,140 +111,22 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
|
|||||||
return tokenString, nil
|
return tokenString, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string]int, platformID int) ([]string, []string, error) {
|
func (a *authDatabase) checkKickToken(ctx context.Context, platformID int, token *tokenverify.Claims) bool {
|
||||||
// todo: Move the logic for handling old data to another location.
|
switch a.multiLoginPolicy {
|
||||||
var (
|
|
||||||
loginTokenMap = make(map[int][]string) // The length of the value of the map must be greater than 0
|
|
||||||
deleteToken = make([]string, 0)
|
|
||||||
kickToken = make([]string, 0)
|
|
||||||
adminToken = make([]string, 0)
|
|
||||||
unkickTerminal = ""
|
|
||||||
)
|
|
||||||
|
|
||||||
for plfID, tks := range tokens {
|
|
||||||
for k, v := range tks {
|
|
||||||
_, err := tokenverify.GetClaimFromToken(k, authverify.Secret(a.accessSecret))
|
|
||||||
if err != nil || v != constant.NormalToken {
|
|
||||||
deleteToken = append(deleteToken, k)
|
|
||||||
} else {
|
|
||||||
if plfID != constant.AdminPlatformID {
|
|
||||||
loginTokenMap[plfID] = append(loginTokenMap[plfID], k)
|
|
||||||
} else {
|
|
||||||
adminToken = append(adminToken, k)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
switch a.multiLogin.Policy {
|
|
||||||
case constant.DefalutNotKick:
|
case constant.DefalutNotKick:
|
||||||
for plt, ts := range loginTokenMap {
|
return false
|
||||||
l := len(ts)
|
|
||||||
if platformID == plt {
|
|
||||||
l++
|
|
||||||
}
|
|
||||||
limit := a.multiLogin.MaxNumOneEnd
|
|
||||||
if l > limit {
|
|
||||||
kickToken = append(kickToken, ts[:l-limit]...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case constant.AllLoginButSameTermKick:
|
|
||||||
for plt, ts := range loginTokenMap {
|
|
||||||
kickToken = append(kickToken, ts[:len(ts)-1]...)
|
|
||||||
if plt == platformID {
|
|
||||||
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:
|
case constant.PCAndOther:
|
||||||
if unkickTerminal == "" {
|
if constant.PlatformIDToClass(platformID) == constant.TerminalPC ||
|
||||||
unkickTerminal = constant.TerminalPC
|
constant.PlatformIDToClass(token.PlatformID) == constant.TerminalPC {
|
||||||
|
return false
|
||||||
}
|
}
|
||||||
if constant.PlatformIDToClass(platformID) != unkickTerminal {
|
return true
|
||||||
for plt, ts := range loginTokenMap {
|
case constant.AllLoginButSameTermKick:
|
||||||
if constant.PlatformIDToClass(plt) != unkickTerminal {
|
if platformID == token.PlatformID {
|
||||||
kickToken = append(kickToken, ts...)
|
return true
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
var (
|
|
||||||
preKick []string
|
|
||||||
isReserve = true
|
|
||||||
)
|
|
||||||
for plt, ts := range loginTokenMap {
|
|
||||||
if constant.PlatformIDToClass(plt) != unkickTerminal {
|
|
||||||
// Keep a token from another end
|
|
||||||
if isReserve {
|
|
||||||
isReserve = false
|
|
||||||
kickToken = append(kickToken, ts[:len(ts)-1]...)
|
|
||||||
preKick = append(preKick, ts[len(ts)-1])
|
|
||||||
continue
|
|
||||||
} else {
|
|
||||||
// Prioritize keeping Android
|
|
||||||
if plt == constant.AndroidPlatformID {
|
|
||||||
kickToken = append(kickToken, preKick...)
|
|
||||||
kickToken = append(kickToken, ts[:len(ts)-1]...)
|
|
||||||
} else {
|
|
||||||
kickToken = append(kickToken, ts...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case constant.PcMobileAndWeb:
|
|
||||||
var (
|
|
||||||
reserved = make(map[string]bool)
|
|
||||||
)
|
|
||||||
|
|
||||||
for plt, ts := range loginTokenMap {
|
|
||||||
if constant.PlatformIDToClass(plt) == constant.PlatformIDToClass(platformID) {
|
|
||||||
kickToken = append(kickToken, ts...)
|
|
||||||
} else {
|
|
||||||
if !reserved[constant.PlatformIDToClass(plt)] {
|
|
||||||
reserved[constant.PlatformIDToClass(plt)] = true
|
|
||||||
kickToken = append(kickToken, ts[:len(ts)-1]...)
|
|
||||||
continue
|
|
||||||
} else {
|
|
||||||
kickToken = append(kickToken, ts...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
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]...)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
return false
|
||||||
default:
|
default:
|
||||||
return nil, nil, errs.New("unknown multiLogin policy").Wrap()
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
|
|
||||||
if a.multiLogin.Policy == constant.Customize {
|
|
||||||
adminTokenMaxNum = a.multiLogin.CustomizeLoginNum[constant.AdminPlatformID]
|
|
||||||
}
|
|
||||||
l := len(adminToken)
|
|
||||||
if platformID == constant.AdminPlatformID {
|
|
||||||
l++
|
|
||||||
}
|
|
||||||
if l > adminTokenMaxNum {
|
|
||||||
kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
|
|
||||||
}
|
|
||||||
return deleteToken, kickToken, nil
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -16,10 +16,9 @@ package controller
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
|
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
|
||||||
@@ -195,7 +194,7 @@ func (c *conversationDatabase) SyncPeerUserPrivateConversationTx(ctx context.Con
|
|||||||
return c.tx.Transaction(ctx, func(ctx context.Context) error {
|
return c.tx.Transaction(ctx, func(ctx context.Context) error {
|
||||||
cache := c.cache.CloneConversationCache()
|
cache := c.cache.CloneConversationCache()
|
||||||
for _, conversation := range conversations {
|
for _, conversation := range conversations {
|
||||||
cache = cache.DelConversationVersionUserIDs(conversation.OwnerUserID, conversation.UserID)
|
cache = cache.DelConversationVersionUserIDs(conversation.OwnerUserID)
|
||||||
for _, v := range [][2]string{{conversation.OwnerUserID, conversation.UserID}, {conversation.UserID, conversation.OwnerUserID}} {
|
for _, v := range [][2]string{{conversation.OwnerUserID, conversation.UserID}, {conversation.UserID, conversation.OwnerUserID}} {
|
||||||
ownerUserID := v[0]
|
ownerUserID := v[0]
|
||||||
userID := v[1]
|
userID := v[1]
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ package controller
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"github.com/openimsdk/protocol/constant"
|
|
||||||
|
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
"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/prommetrics"
|
||||||
@@ -24,11 +23,8 @@ type MsgTransferDatabase interface {
|
|||||||
DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
|
DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
|
||||||
|
|
||||||
// BatchInsertChat2Cache increments the sequence number and then batch inserts messages into the cache.
|
// BatchInsertChat2Cache increments the sequence number and then batch inserts messages into the cache.
|
||||||
BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, userHasReadMap map[string]int64, err error)
|
BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error)
|
||||||
|
SetHasReadSeqToDB(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
|
||||||
SetHasReadSeqs(ctx context.Context, conversationID string, userSeqMap map[string]int64) error
|
|
||||||
|
|
||||||
SetHasReadSeqToDB(ctx context.Context, conversationID string, userSeqMap map[string]int64) error
|
|
||||||
|
|
||||||
// to mq
|
// to mq
|
||||||
MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
|
MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
|
||||||
@@ -87,9 +83,6 @@ func (db *msgTransferDatabase) BatchInsertChat2DB(ctx context.Context, conversat
|
|||||||
IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
|
IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if msg.Status == constant.MsgStatusSending {
|
|
||||||
msg.Status = constant.MsgStatusSendSuccess
|
|
||||||
}
|
|
||||||
msgs[i] = &model.MsgDataModel{
|
msgs[i] = &model.MsgDataModel{
|
||||||
SendID: msg.SendID,
|
SendID: msg.SendID,
|
||||||
RecvID: msg.RecvID,
|
RecvID: msg.RecvID,
|
||||||
@@ -222,18 +215,18 @@ func (db *msgTransferDatabase) DeleteMessagesFromCache(ctx context.Context, conv
|
|||||||
return db.msg.DeleteMessagesFromCache(ctx, conversationID, seqs)
|
return db.msg.DeleteMessagesFromCache(ctx, conversationID, seqs)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (db *msgTransferDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, userHasReadMap map[string]int64, err error) {
|
func (db *msgTransferDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
|
||||||
lenList := len(msgs)
|
lenList := len(msgs)
|
||||||
if int64(lenList) > db.msgTable.GetSingleGocMsgNum() {
|
if int64(lenList) > db.msgTable.GetSingleGocMsgNum() {
|
||||||
return 0, false, nil, errs.New("message count exceeds limit", "limit", db.msgTable.GetSingleGocMsgNum()).Wrap()
|
return 0, false, errs.New("message count exceeds limit", "limit", db.msgTable.GetSingleGocMsgNum()).Wrap()
|
||||||
}
|
}
|
||||||
if lenList < 1 {
|
if lenList < 1 {
|
||||||
return 0, false, nil, errs.New("no messages to insert", "minCount", 1).Wrap()
|
return 0, false, errs.New("no messages to insert", "minCount", 1).Wrap()
|
||||||
}
|
}
|
||||||
currentMaxSeq, err := db.seqConversation.Malloc(ctx, conversationID, int64(len(msgs)))
|
currentMaxSeq, err := db.seqConversation.Malloc(ctx, conversationID, int64(len(msgs)))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.ZError(ctx, "storage.seq.Malloc", err)
|
log.ZError(ctx, "storage.seq.Malloc", err)
|
||||||
return 0, false, nil, err
|
return 0, false, err
|
||||||
}
|
}
|
||||||
isNew = currentMaxSeq == 0
|
isNew = currentMaxSeq == 0
|
||||||
lastMaxSeq := currentMaxSeq
|
lastMaxSeq := currentMaxSeq
|
||||||
@@ -251,10 +244,15 @@ func (db *msgTransferDatabase) BatchInsertChat2Cache(ctx context.Context, conver
|
|||||||
} else {
|
} else {
|
||||||
prommetrics.MsgInsertRedisSuccessCounter.Inc()
|
prommetrics.MsgInsertRedisSuccessCounter.Inc()
|
||||||
}
|
}
|
||||||
return lastMaxSeq, isNew, userSeqMap, errs.Wrap(err)
|
err = db.setHasReadSeqs(ctx, conversationID, userSeqMap)
|
||||||
|
if err != nil {
|
||||||
|
log.ZError(ctx, "SetHasReadSeqs error", err, "userSeqMap", userSeqMap, "conversationID", conversationID)
|
||||||
|
prommetrics.SeqSetFailedCounter.Inc()
|
||||||
|
}
|
||||||
|
return lastMaxSeq, isNew, errs.Wrap(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (db *msgTransferDatabase) SetHasReadSeqs(ctx context.Context, conversationID string, userSeqMap map[string]int64) error {
|
func (db *msgTransferDatabase) setHasReadSeqs(ctx context.Context, conversationID string, userSeqMap map[string]int64) error {
|
||||||
for userID, seq := range userSeqMap {
|
for userID, seq := range userSeqMap {
|
||||||
if err := db.seqUser.SetUserReadSeq(ctx, conversationID, userID, seq); err != nil {
|
if err := db.seqUser.SetUserReadSeq(ctx, conversationID, userID, seq); err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -263,13 +261,8 @@ func (db *msgTransferDatabase) SetHasReadSeqs(ctx context.Context, conversationI
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (db *msgTransferDatabase) SetHasReadSeqToDB(ctx context.Context, conversationID string, userSeqMap map[string]int64) error {
|
func (db *msgTransferDatabase) SetHasReadSeqToDB(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error {
|
||||||
for userID, seq := range userSeqMap {
|
return db.seqUser.SetUserReadSeqToDB(ctx, conversationID, userID, hasReadSeq)
|
||||||
if err := db.seqUser.SetUserReadSeqToDB(ctx, conversationID, userID, seq); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (db *msgTransferDatabase) MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error) {
|
func (db *msgTransferDatabase) MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error) {
|
||||||
|
|||||||
@@ -16,7 +16,6 @@ package rpccache
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
"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/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
|
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
|
||||||
@@ -98,7 +97,6 @@ func (u *UserLocalCache) GetUsersInfo(ctx context.Context, userIDs []string) ([]
|
|||||||
user, err := u.GetUserInfo(ctx, userID)
|
user, err := u.GetUserInfo(ctx, userID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errs.ErrRecordNotFound.Is(err) {
|
if errs.ErrRecordNotFound.Is(err) {
|
||||||
log.ZWarn(ctx, "User info notFound", err, "userID", userID)
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|||||||
+5
-16
@@ -16,8 +16,8 @@ package rpcclient
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
|
||||||
"github.com/openimsdk/protocol/auth"
|
"github.com/openimsdk/protocol/auth"
|
||||||
|
pbAuth "github.com/openimsdk/protocol/auth"
|
||||||
"github.com/openimsdk/tools/discovery"
|
"github.com/openimsdk/tools/discovery"
|
||||||
"github.com/openimsdk/tools/system/program"
|
"github.com/openimsdk/tools/system/program"
|
||||||
"google.golang.org/grpc"
|
"google.golang.org/grpc"
|
||||||
@@ -38,8 +38,8 @@ type Auth struct {
|
|||||||
discov discovery.SvcDiscoveryRegistry
|
discov discovery.SvcDiscoveryRegistry
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *Auth) ParseToken(ctx context.Context, token string) (*auth.ParseTokenResp, error) {
|
func (a *Auth) ParseToken(ctx context.Context, token string) (*pbAuth.ParseTokenResp, error) {
|
||||||
req := auth.ParseTokenReq{
|
req := pbAuth.ParseTokenReq{
|
||||||
Token: token,
|
Token: token,
|
||||||
}
|
}
|
||||||
resp, err := a.Client.ParseToken(ctx, &req)
|
resp, err := a.Client.ParseToken(ctx, &req)
|
||||||
@@ -49,8 +49,8 @@ func (a *Auth) ParseToken(ctx context.Context, token string) (*auth.ParseTokenRe
|
|||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *Auth) InvalidateToken(ctx context.Context, preservedToken, userID string, platformID int) (*auth.InvalidateTokenResp, error) {
|
func (a *Auth) InvalidateToken(ctx context.Context, preservedToken, userID string, platformID int) (*pbAuth.InvalidateTokenResp, error) {
|
||||||
req := auth.InvalidateTokenReq{
|
req := pbAuth.InvalidateTokenReq{
|
||||||
PreservedToken: preservedToken,
|
PreservedToken: preservedToken,
|
||||||
UserID: userID,
|
UserID: userID,
|
||||||
PlatformID: int32(platformID),
|
PlatformID: int32(platformID),
|
||||||
@@ -61,14 +61,3 @@ func (a *Auth) InvalidateToken(ctx context.Context, preservedToken, userID strin
|
|||||||
}
|
}
|
||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *Auth) KickTokens(ctx context.Context, tokens []string) (*auth.KickTokensResp, error) {
|
|
||||||
req := auth.KickTokensReq{
|
|
||||||
Tokens: tokens,
|
|
||||||
}
|
|
||||||
resp, err := a.Client.KickTokens(ctx, &req)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return resp, err
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user