Compare commits

...

4 Commits

Author SHA1 Message Date
github-actions[bot]
a2677e17c4
Merge ddd16affeab698add80a1db3e5b13ab8f7c74df3 into 1178808ba7d7798cfc13aada0b47c4a1e130cdce 2025-05-14 15:25:54 +08:00
chao
1178808ba7
feat: optimize server code (#3319)
* pb

* fix: Modifying other fields while setting IsPrivateChat does not take effect

* fix: quote message error revoke

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* refactoring scheduled tasks

* upgrading pkg tools

* fix

* fix

* optimize log output

* feat: support GetLastMessage

* feat: support GetLastMessage

* feat: s3 switch

* feat: s3 switch

* fix: GetUsersOnline

* feat: SendBusinessNotification supported configuration parameters

* feat: SendBusinessNotification supported configuration parameters

* feat: SendBusinessNotification supported configuration parameters

* feat: seq conversion failed without exiting

* fix: DeleteDoc crash

* fix: fill send time

* fix: fill send time

* fix: crash caused by withdrawing messages from users who have left the group

* fix: user msg timestamp

* seq read config

* seq read config

* fix: the source message of the reference is withdrawn, and the referenced message is deleted

* feat: optimize the default notification.yml

* fix: shouldPushOffline

* fix: the sorting is wrong after canceling the administrator in group settings

* feat: Sending messages supports returning fields modified by webhook

* feat: Sending messages supports returning fields modified by webhook

* feat: Sending messages supports returning fields modified by webhook

* fix: oss specifies content-type when uploading

* fix: the version number contains a line break

* fix: the version number contains a line break

* feat: GetConversationsHasReadAndMaxSeq support pinned

* feat: GetConversationsHasReadAndMaxSeq support pinned

* feat: GetConversationsHasReadAndMaxSeq support pinned

* fix: transferring the group owner to a muted member, incremental version error

* feat: unified conversion code

* feat: update gomake
2025-05-09 01:31:30 +00:00
icey-yu
56c5c1f015
fix: delete token by correct platformID && feat: adminToken can be retained for five minutes after deleting (#3313) 2025-05-06 07:10:10 +00:00
mo3et
ddd16affea Update CHANGELOG for release v3.8.3-patch.4 2025-03-13 08:21:22 +00:00
12 changed files with 192 additions and 81 deletions

View File

@ -0,0 +1,7 @@
## [v3.8.3-patch.4](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.4) (2025-03-13)
### Bug Fixes
* fix: solve unocrrect invite notificationfrom #3213
**Full Changelog**: [v3.8.3-patch.3...v3.8.3-patch.4](https://github.com/openimsdk/open-im-server/compare/v3.8.3-patch.3...v3.8.3-patch.4)

4
CHANGELOG/README.md Normal file
View File

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

2
go.mod
View File

@ -34,7 +34,7 @@ require (
github.com/hashicorp/golang-lru/v2 v2.0.7
github.com/kelindar/bitmap v1.5.2
github.com/likexian/gokit v0.25.13
github.com/openimsdk/gomake v0.0.15-alpha.2
github.com/openimsdk/gomake v0.0.15-alpha.5
github.com/redis/go-redis/v9 v9.4.0
github.com/robfig/cron/v3 v3.0.1
github.com/shirou/gopsutil v3.21.11+incompatible

4
go.sum
View File

@ -345,8 +345,8 @@ github.com/onsi/ginkgo/v2 v2.19.0 h1:9Cnnf7UHo57Hy3k6/m5k3dRfGTMXGvxhHFvkDTCTpvA
github.com/onsi/ginkgo/v2 v2.19.0/go.mod h1:rlwLi9PilAFJ8jCg9UE1QP6VBpd6/xj3SRC0d6TU0To=
github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM=
github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/gomake v0.0.15-alpha.5 h1:eEZCEHm+NsmcO3onXZPIUbGFCYPYbsX5beV3ZyOsGhY=
github.com/openimsdk/gomake v0.0.15-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/protocol v0.0.73-alpha.6 h1:sna9coWG7HN1zObBPtvG0Ki/vzqHXiB4qKbA5P3w7kc=
github.com/openimsdk/protocol v0.0.73-alpha.6/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/tools v0.0.50-alpha.79 h1:jxYEbrzaze4Z2r4NrKad816buZ690ix0L9MTOOOH3ik=

View File

@ -159,15 +159,17 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim
if err != nil {
return nil, err
}
isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
if isAdmin {
return claims, nil
}
m, err := s.authDatabase.GetTokensWithoutError(ctx, claims.UserID, claims.PlatformID)
if err != nil {
return nil, err
}
if len(m) == 0 {
isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
if isAdmin {
if err = s.authDatabase.GetTemporaryTokensWithoutError(ctx, claims.UserID, claims.PlatformID, tokensString); err == nil {
return claims, nil
}
}
return nil, servererrs.ErrTokenNotExist.Wrap()
}
if v, ok := m[tokensString]; ok {
@ -179,6 +181,13 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim
default:
return nil, errs.Wrap(errs.ErrTokenUnknown)
}
} else {
isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
if isAdmin {
if err = s.authDatabase.GetTemporaryTokensWithoutError(ctx, claims.UserID, claims.PlatformID, tokensString); err == nil {
return claims, nil
}
}
}
return nil, servererrs.ErrTokenNotExist.Wrap()
}

View File

@ -1,8 +1,9 @@
package cachekey
import (
"github.com/openimsdk/protocol/constant"
"strings"
"github.com/openimsdk/protocol/constant"
)
const (
@ -13,6 +14,10 @@ func GetTokenKey(userID string, platformID int) string {
return UidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
}
func GetTemporaryTokenKey(userID string, platformID int, token string) string {
return UidPidToken + ":TEMPORARY:" + userID + ":" + constant.PlatformIDToName(platformID) + ":" + token
}
func GetAllPlatformTokenKey(userID string) []string {
res := make([]string, len(constant.PlatformID2Name))
for k := range constant.PlatformID2Name {

View File

@ -27,7 +27,6 @@ type tokenCache struct {
func (x *tokenCache) getTokenKey(userID string, platformID int, token string) string {
return cachekey.GetTokenKey(userID, platformID) + ":" + token
}
func (x *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
@ -57,6 +56,14 @@ func (x *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
return mm, nil
}
func (x *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error {
key := cachekey.GetTemporaryTokenKey(userID, platformID, token)
if _, err := x.cache.Get(ctx, []string{key}); err != nil {
return err
}
return nil
}
func (x *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
prefix := cachekey.UidPidToken + userID + ":"
tokens, err := x.cache.Prefix(ctx, prefix)
@ -128,3 +135,32 @@ func (x *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, pla
func (x *tokenCache) getExpireTime(t int64) time.Duration {
return time.Hour * 24 * time.Duration(t)
}
func (x *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error {
keys := make([]string, 0, len(tokens))
for platformID, ts := range tokens {
for _, t := range ts {
keys = append(keys, x.getTokenKey(userID, platformID, t))
}
}
return x.cache.Del(ctx, keys)
}
func (x *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error {
keys := make([]string, 0, len(fields))
for _, f := range fields {
keys = append(keys, x.getTokenKey(userID, platformID, f))
}
if err := x.cache.Del(ctx, keys); err != nil {
return err
}
for _, f := range fields {
k := cachekey.GetTemporaryTokenKey(userID, platformID, f)
if err := x.cache.Set(ctx, k, "", time.Minute*5); err != nil {
return errs.Wrap(err)
}
}
return nil
}

View File

@ -9,6 +9,7 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
"github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/utils/datautil"
"github.com/redis/go-redis/v9"
)
@ -55,6 +56,14 @@ func (c *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
return mm, nil
}
func (c *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error {
err := c.rdb.Get(ctx, cachekey.GetTemporaryTokenKey(userID, platformID, token)).Err()
if err != nil {
return errs.Wrap(err)
}
return nil
}
func (c *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
var (
res = make(map[int]map[string]int)
@ -101,13 +110,19 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla
}
func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) 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)
keys := datautil.Keys(tokens)
if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) 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
}); err != nil {
return err
}
return nil
}
@ -119,3 +134,47 @@ func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, pla
func (c *tokenCache) getExpireTime(t int64) time.Duration {
return time.Hour * 24 * time.Duration(t)
}
// DeleteTokenByTokenMap tokens key is platformID, value is token slice
func (c *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error {
var (
keys = make([]string, 0, len(tokens))
keyMap = make(map[string][]string)
)
for k, v := range tokens {
k1 := cachekey.GetTokenKey(userID, k)
keys = append(keys, k1)
keyMap[k1] = v
}
if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
pipe := c.rdb.Pipeline()
for k, v := range tokens {
pipe.HDel(ctx, cachekey.GetTokenKey(userID, k), v...)
}
_, err := pipe.Exec(ctx)
if err != nil {
return errs.Wrap(err)
}
return nil
}); err != nil {
return err
}
return nil
}
func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error {
key := cachekey.GetTokenKey(userID, platformID)
if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
return errs.Wrap(err)
}
for _, f := range fields {
k := cachekey.GetTemporaryTokenKey(userID, platformID, f)
if err := c.rdb.Set(ctx, k, "", time.Minute*5).Err(); err != nil {
return errs.Wrap(err)
}
}
return nil
}

View File

@ -9,8 +9,11 @@ type TokenModel interface {
// SetTokenFlagEx set token and flag with expire time
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)
HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) 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
BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error
DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error
DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error
}

View File

@ -17,6 +17,8 @@ import (
type AuthDatabase interface {
// If the result is empty, no error is returned.
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
GetTemporaryTokensWithoutError(ctx context.Context, userID string, platformID int, token string) error
// Create token
CreateToken(ctx context.Context, userID string, platformID int) (string, error)
@ -52,6 +54,10 @@ func (a *authDatabase) GetTokensWithoutError(ctx context.Context, userID string,
return a.cache.GetTokensWithoutError(ctx, userID, platformID)
}
func (a *authDatabase) GetTemporaryTokensWithoutError(ctx context.Context, userID string, platformID int, token string) error {
return a.cache.HasTemporaryToken(ctx, userID, platformID, token)
}
func (a *authDatabase) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
return a.cache.SetTokenMapByUidPid(ctx, userID, platformID, m)
}
@ -85,23 +91,30 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
return "", err
}
deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
deleteTokenKey, kickedTokenKey, adminTokens, err := a.checkToken(ctx, tokens, platformID)
if err != nil {
return "", err
}
if len(deleteTokenKey) != 0 {
err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
err = a.cache.DeleteTokenByTokenMap(ctx, userID, deleteTokenKey)
if err != nil {
return "", err
}
}
if len(kickedTokenKey) != 0 {
for _, k := range kickedTokenKey {
err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
if err != nil {
return "", err
for plt, ks := range kickedTokenKey {
for _, k := range ks {
err := a.cache.SetTokenFlagEx(ctx, userID, plt, k, constant.KickedToken)
if err != nil {
return "", err
}
log.ZDebug(ctx, "kicked token in create token", "token", k)
}
log.ZDebug(ctx, "kicked token in create token", "token", k)
}
}
if len(adminTokens) != 0 {
if err = a.cache.DeleteAndSetTemporary(ctx, userID, constant.AdminPlatformID, adminTokens); err != nil {
return "", err
}
}
@ -119,12 +132,13 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
return tokenString, nil
}
func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string]int, platformID int) ([]string, []string, error) {
// todo: Move the logic for handling old data to another location.
// checkToken will check token by tokenPolicy and return deleteToken,kickToken,deleteAdminToken
func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string]int, platformID int) (map[int][]string, map[int][]string, []string, error) {
// todo: Asynchronous deletion of old data.
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)
deleteToken = make(map[int][]string)
kickToken = make(map[int][]string)
adminToken = make([]string, 0)
unkickTerminal = ""
)
@ -133,7 +147,7 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
for k, v := range tks {
_, err := tokenverify.GetClaimFromToken(k, authverify.Secret(a.accessSecret))
if err != nil || v != constant.NormalToken {
deleteToken = append(deleteToken, k)
deleteToken[plfID] = append(deleteToken[plfID], k)
} else {
if plfID != constant.AdminPlatformID {
loginTokenMap[plfID] = append(loginTokenMap[plfID], k)
@ -153,14 +167,15 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
}
limit := a.multiLogin.MaxNumOneEnd
if l > limit {
kickToken = append(kickToken, ts[:l-limit]...)
kickToken[plt] = ts[:l-limit]
}
}
case constant.AllLoginButSameTermKick:
for plt, ts := range loginTokenMap {
kickToken = append(kickToken, ts[:len(ts)-1]...)
kickToken[plt] = ts[:len(ts)-1]
if plt == platformID {
kickToken = append(kickToken, ts[len(ts)-1])
kickToken[plt] = append(kickToken[plt], ts[len(ts)-1])
}
}
case constant.PCAndOther:
@ -168,29 +183,33 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
if constant.PlatformIDToClass(platformID) != unkickTerminal {
for plt, ts := range loginTokenMap {
if constant.PlatformIDToClass(plt) != unkickTerminal {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
}
}
} else {
var (
preKick []string
isReserve = true
preKickToken string
preKickPlt int
reserveToken = false
)
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])
if !reserveToken {
reserveToken = true
kickToken[plt] = ts[:len(ts)-1]
preKickToken = ts[len(ts)-1]
preKickPlt = plt
continue
} else {
// Prioritize keeping Android
if plt == constant.AndroidPlatformID {
kickToken = append(kickToken, preKick...)
kickToken = append(kickToken, ts[:len(ts)-1]...)
if preKickToken != "" {
kickToken[preKickPlt] = append(kickToken[preKickPlt], preKickToken)
}
kickToken[plt] = ts[:len(ts)-1]
} else {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
}
}
}
@ -203,19 +222,19 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
for plt, ts := range loginTokenMap {
if constant.PlatformIDToClass(plt) == constant.PlatformIDToClass(platformID) {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
} else {
if _, ok := reserved[constant.PlatformIDToClass(plt)]; !ok {
reserved[constant.PlatformIDToClass(plt)] = struct{}{}
kickToken = append(kickToken, ts[:len(ts)-1]...)
kickToken[plt] = ts[:len(ts)-1]
continue
} else {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
}
}
}
default:
return nil, nil, errs.New("unknown multiLogin policy").Wrap()
return nil, nil, nil, errs.New("unknown multiLogin policy").Wrap()
}
//var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
@ -226,8 +245,9 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
//if l > adminTokenMaxNum {
// kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
//}
var deleteAdminToken []string
if platformID == constant.AdminPlatformID {
kickToken = append(kickToken, adminToken...)
deleteAdminToken = adminToken
}
return deleteToken, kickToken, nil
return deleteToken, kickToken, deleteAdminToken, nil
}

View File

@ -81,42 +81,10 @@ func (db *msgTransferDatabase) BatchInsertChat2DB(ctx context.Context, conversat
continue
}
seqs[i] = msg.Seq
var offlinePushModel *model.OfflinePushModel
if msg.OfflinePushInfo != nil {
offlinePushModel = &model.OfflinePushModel{
Title: msg.OfflinePushInfo.Title,
Desc: msg.OfflinePushInfo.Desc,
Ex: msg.OfflinePushInfo.Ex,
IOSPushSound: msg.OfflinePushInfo.IOSPushSound,
IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
}
}
if msg.Status == constant.MsgStatusSending {
msg.Status = constant.MsgStatusSendSuccess
}
msgs[i] = &model.MsgDataModel{
SendID: msg.SendID,
RecvID: msg.RecvID,
GroupID: msg.GroupID,
ClientMsgID: msg.ClientMsgID,
ServerMsgID: msg.ServerMsgID,
SenderPlatformID: msg.SenderPlatformID,
SenderNickname: msg.SenderNickname,
SenderFaceURL: msg.SenderFaceURL,
SessionType: msg.SessionType,
MsgFrom: msg.MsgFrom,
ContentType: msg.ContentType,
Content: string(msg.Content),
Seq: msg.Seq,
SendTime: msg.SendTime,
CreateTime: msg.CreateTime,
Status: msg.Status,
Options: msg.Options,
OfflinePush: offlinePushModel,
AtUserIDList: msg.AtUserIDList,
AttachedInfo: msg.AttachedInfo,
Ex: msg.Ex,
}
msgs[i] = convert.MsgPb2DB(msg)
}
if err := db.BatchInsertBlock(ctx, conversationID, msgs, updateKeyMsg, msgList[0].Seq); err != nil {
return err

View File

@ -127,7 +127,7 @@ func (x *CacheMgo) Del(ctx context.Context, key []string) error {
return nil
}
_, err := x.coll.DeleteMany(ctx, bson.M{"key": bson.M{"$in": key}})
return err
return errs.Wrap(err)
}
func (x *CacheMgo) lockKey(key string) string {