Merge pull request #3174 from icey-yu/fix-push

fix: Offline push does not have a badge && Android offline push (#3146)

.env

@@ -17,8 +17,6 @@ OPENIM_ADMIN_FRONT_IMAGE=openim/openim-admin-front:release-v1.8.4
 
 DATA_DIR=./
 
-MONGO_BACKUP_DIR=${DATA_DIR}components/backup/mongo/
-
 PROMETHEUS_PORT=19091
 ALERTMANAGER_PORT=19093
 GRAFANA_PORT=13000

docker-compose.yml

@@ -37,7 +37,6 @@ services:
       - "${DATA_DIR}/components/mongodb/data/db:/data/db"
       - "${DATA_DIR}/components/mongodb/data/logs:/data/logs"
       - "${DATA_DIR}/components/mongodb/data/conf:/etc/mongo"
-      - "${MONGO_BACKUP_DIR}:/data/backup"
     environment:
       - TZ=Asia/Shanghai
       - wiredTigerCacheSizeGB=1

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/openimsdk/protocol v0.0.72-alpha.71
-	github.com/openimsdk/tools v0.0.50-alpha.72
+	github.com/openimsdk/tools v0.0.50-alpha.67
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0

go.sum

@@ -321,8 +321,8 @@ github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCF
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
 github.com/openimsdk/protocol v0.0.72-alpha.71 h1:R3utzOlqepaJWTAmnfJi4ccUM/XIoFasSyjQMOipM70=
 github.com/openimsdk/protocol v0.0.72-alpha.71/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
-github.com/openimsdk/tools v0.0.50-alpha.72 h1:d/vaZjIfvrNp3EeRJEIiamBO7HiPx6CP4wiuq8NpfzY=
-github.com/openimsdk/tools v0.0.50-alpha.72/go.mod h1:B+oqV0zdewN7OiEHYJm+hW+8/Te7B8tHHgD8rK5ZLZk=
+github.com/openimsdk/tools v0.0.50-alpha.67 h1:K7kguqvPbjldHAi7pGhcG2ERkctCqG9ZFlteT7UKaxM=
+github.com/openimsdk/tools v0.0.50-alpha.67/go.mod h1:B+oqV0zdewN7OiEHYJm+hW+8/Te7B8tHHgD8rK5ZLZk=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

internal/push/offlinepush/fcm/push.go

@@ -16,12 +16,14 @@ package fcm
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
-	"github.com/openimsdk/tools/utils/httputil"
 	"path/filepath"
 	"strings"
 
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
+	"github.com/openimsdk/tools/utils/httputil"
+
 	firebase "firebase.google.com/go/v4"
 	"firebase.google.com/go/v4/messaging"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -133,7 +135,7 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
 			unreadCountSum, err := f.cache.GetUserBadgeUnreadCountSum(ctx, userID)
 			if err == nil && unreadCountSum != 0 {
 				apns.Payload.Aps.Badge = &unreadCountSum
-			} else if err == redis.Nil || unreadCountSum == 0 {
+			} else if errors.Is(err, redis.Nil) || unreadCountSum == 0 {
 				zero := 1
 				apns.Payload.Aps.Badge = &zero
 			} else {

internal/push/offlinepush/getui/body.go

@@ -18,6 +18,16 @@ import (
 	"fmt"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/tools/utils/datautil"
+)
+
+var (
+	incOne          = datautil.ToPtr("+1")
+	addNum          = "1"
+	defaultStrategy = strategy{
+		Default: 1,
+	}
+	msgCategory = "CATEGORY_MESSAGE"
 )
 
 type Resp struct {
@@ -58,7 +68,24 @@ type TaskResp struct {
 }
 
 type Settings struct {
-	TTL *int64 `json:"ttl"`
+	TTL      *int64   `json:"ttl"`
+	Strategy strategy `json:"strategy"`
+}
+
+type strategy struct {
+	Default int64 `json:"default"`
+	//IOS     int64 `json:"ios"`
+	//St      int64 `json:"st"`
+	//Hw      int64 `json:"hw"`
+	//Ho      int64 `json:"ho"`
+	//XM      int64 `json:"xm"`
+	//XMG     int64 `json:"xmg"`
+	//VV      int64 `json:"vv"`
+	//Op      int64 `json:"op"`
+	//OpG     int64 `json:"opg"`
+	//MZ      int64 `json:"mz"`
+	//HosHw   int64 `json:"hoshw"`
+	//WX      int64 `json:"wx"`
 }
 
 type Audience struct {
@@ -112,6 +139,8 @@ type Notification struct {
 	ChannelID   string `json:"channelID"`
 	ChannelName string `json:"ChannelName"`
 	ClickType   string `json:"click_type"`
+	BadgeAddNum string `json:"badge_add_num"`
+	Category    string `json:"category"`
 }
 
 type Options struct {
@@ -120,6 +149,7 @@ type Options struct {
 		ChannelID    string `json:"/message/android/notification/channel_id"`
 		Sound        string `json:"/message/android/notification/sound"`
 		Importance   string `json:"/message/android/notification/importance"`
+		Category     string `json:"/message/android/category"`
 	} `json:"HW"`
 	XM struct {
 		ChannelID string `json:"/extra.channel_id"`
@@ -140,6 +170,8 @@ func newPushReq(pushConf *config.Push, title, content string) PushReq {
 		ClickType:   "startapp",
 		ChannelID:   pushConf.GeTui.ChannelID,
 		ChannelName: pushConf.GeTui.ChannelName,
+		BadgeAddNum: addNum,
+		Category:    msgCategory,
 	}}}
 	return pushReq
 }
@@ -156,6 +188,7 @@ func (pushReq *PushReq) setPushChannel(title string, body string) {
 	notify := "notify"
 	pushReq.PushChannel.Ios.NotificationType = &notify
 	pushReq.PushChannel.Ios.Aps.Sound = "default"
+	pushReq.PushChannel.Ios.AutoBadge = incOne
 	pushReq.PushChannel.Ios.Aps.Alert = Alert{
 		Title: title,
 		Body:  body,
@@ -172,7 +205,8 @@ func (pushReq *PushReq) setPushChannel(title string, body string) {
 			ChannelID    string `json:"/message/android/notification/channel_id"`
 			Sound        string `json:"/message/android/notification/sound"`
 			Importance   string `json:"/message/android/notification/importance"`
-		}{ChannelID: "RingRing4", Sound: "/raw/ring001", Importance: "NORMAL"},
+			Category     string `json:"/message/android/category"`
+		}{ChannelID: "RingRing4", Sound: "/raw/ring001", Importance: "NORMAL", Category: "IM"},
 		XM: struct {
 			ChannelID string `json:"/extra.channel_id"`
 		}{ChannelID: "high_system"},

internal/push/offlinepush/getui/push.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"strconv"
 	"sync"
 	"time"
@@ -70,7 +71,7 @@ func NewClient(pushConf *config.Push, cache cache.ThirdCache) *Client {
 func (g *Client) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
 	token, err := g.cache.GetGetuiToken(ctx)
 	if err != nil {
-		if errs.Unwrap(err) == redis.Nil {
+		if errors.Is(err, redis.Nil) {
 			log.ZDebug(ctx, "getui token not exist in redis")
 			token, err = g.getTokenAndSave2Redis(ctx)
 			if err != nil {
@@ -144,7 +145,7 @@ func (g *Client) Auth(ctx context.Context, timeStamp int64) (token string, expir
 func (g *Client) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
 	respTask := TaskResp{}
 	ttl := int64(1000 * 60 * 5)
-	pushReq.Settings = &Settings{TTL: &ttl}
+	pushReq.Settings = &Settings{TTL: &ttl, Strategy: defaultStrategy}
 	err := g.request(ctx, taskURL, pushReq, token, &respTask)
 	if err != nil {
 		return "", errs.Wrap(err)
@@ -188,6 +189,7 @@ func (g *Client) postReturn(
 	if err != nil {
 		return err
 	}
+	log.ZDebug(ctx, "postReturn", "url", url, "header", header, "input", input, "timeout", timeout, "output", output)
 	return output.parseError()
 }
 
@@ -204,7 +206,7 @@ func (g *Client) getTokenAndSave2Redis(ctx context.Context) (token string, err e
 }
 
 func (g *Client) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
-	pushReq.Settings = &Settings{TTL: &g.taskIDTTL}
+	pushReq.Settings = &Settings{TTL: &g.taskIDTTL, Strategy: defaultStrategy}
 	taskID, err = g.GetTaskID(ctx, token, pushReq)
 	if err != nil {
 		return

internal/push/push_handler.go

@@ -3,12 +3,11 @@ package push
 import (
 	"context"
 	"encoding/json"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"math/rand"
 	"strconv"
 	"time"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
-
 	"github.com/IBM/sarama"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
@@ -153,24 +152,24 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *
 	log.ZInfo(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
 	defer func(duration time.Time) {
 		t := time.Since(duration)
-		log.ZInfo(ctx, "Get msg from msg_transfer And push msg end", "msg", msg.String(), "time cost", t)
+		log.ZInfo(ctx, "Get msg from msg_transfer And push msg", "msg", msg.String(), "time cost", t)
 	}(time.Now())
 	if err := c.webhookBeforeOnlinePush(ctx, &c.config.WebhooksConfig.BeforeOnlinePush, userIDs, msg); err != nil {
 		return err
 	}
+	log.ZInfo(ctx, "webhookBeforeOnlinePush end")
 
 	wsResults, err := c.GetConnsAndOnlinePush(ctx, msg, userIDs)
 	if err != nil {
 		return err
 	}
 
-	log.ZDebug(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
-	log.ZInfo(ctx, "single and notification push end")
+	log.ZInfo(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
 
 	if !c.shouldPushOffline(ctx, msg) {
 		return nil
 	}
-	log.ZInfo(ctx, "pushOffline start")
+	log.ZInfo(ctx, "shouldPushOffline end")
 
 	for _, v := range wsResults {
 		//message sender do not need offline push
@@ -189,14 +188,14 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *
 	if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserID, msg, &offlinePushUserID); err != nil {
 		return err
 	}
+	log.ZInfo(ctx, "webhookBeforeOfflinePush end")
 
 	if len(offlinePushUserID) > 0 {
 		needOfflinePushUserID = offlinePushUserID
 	}
 	err = c.offlinePushMsg(ctx, msg, needOfflinePushUserID)
 	if err != nil {
-		log.ZDebug(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID", needOfflinePushUserID, "msg", msg)
-		log.ZWarn(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID length", len(needOfflinePushUserID), "msg", msg)
+		log.ZWarn(ctx, "offlinePushMsg failed", err, "needOfflinePushUserID", needOfflinePushUserID, "msg", msg)
 		return nil
 	}
 
@@ -251,24 +250,26 @@ func (c *ConsumerHandler) Push2Group(ctx context.Context, groupID string, msg *s
 		&pushToUserIDs); err != nil {
 		return err
 	}
+	log.ZInfo(ctx, "webhookBeforeGroupOnlinePush end")
 
 	err = c.groupMessagesHandler(ctx, groupID, &pushToUserIDs, msg)
 	if err != nil {
 		return err
 	}
+	log.ZInfo(ctx, "groupMessagesHandler end")
 
 	wsResults, err := c.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
 	if err != nil {
 		return err
 	}
 
-	log.ZDebug(ctx, "group push result", "result", wsResults, "msg", msg)
-	log.ZInfo(ctx, "online group push end")
+	log.ZInfo(ctx, "group push result", "result", wsResults, "msg", msg)
 
 	if !c.shouldPushOffline(ctx, msg) {
 		return nil
 	}
 	needOfflinePushUserIDs := c.onlinePusher.GetOnlinePushFailedUserIDs(ctx, msg, wsResults, &pushToUserIDs)
+	log.ZInfo(ctx, "GetOnlinePushFailedUserIDs end")
 	//filter some user, like don not disturb or don't need offline push etc.
 	needOfflinePushUserIDs, err = c.filterGroupMessageOfflinePush(ctx, groupID, msg, needOfflinePushUserIDs)
 	if err != nil {
@@ -296,11 +297,9 @@ func (c *ConsumerHandler) asyncOfflinePush(ctx context.Context, needOfflinePushU
 		needOfflinePushUserIDs = offlinePushUserIDs
 	}
 	if err := c.pushDatabase.MsgToOfflinePushMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(msg.SendID, msg.RecvID), needOfflinePushUserIDs, msg); err != nil {
-		log.ZDebug(ctx, "Msg To OfflinePush MQ error", err, "needOfflinePushUserIDs",
+		log.ZError(ctx, "Msg To OfflinePush MQ error", err, "needOfflinePushUserIDs",
 			needOfflinePushUserIDs, "msg", msg)
-		log.ZWarn(ctx, "Msg To OfflinePush MQ error", err, "needOfflinePushUserIDs length",
-			len(needOfflinePushUserIDs), "msg", msg)
-		prommetrics.GroupChatMsgProcessFailedCounter.Inc()
+		prommetrics.SingleChatMsgProcessFailedCounter.Inc()
 		return
 	}
 }

internal/rpc/conversation/conversation.go

@@ -16,11 +16,10 @@ package conversation
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"sort"
 	"time"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
@@ -772,7 +771,7 @@ func (c *conversationServer) ClearUserConversationMsg(ctx context.Context, req *
 		if conversation.IsMsgDestruct == false || conversation.MsgDestructTime == 0 {
 			continue
 		}
-		seq, err := c.msgClient.GetLastMessageSeqByTime(ctx, conversation.ConversationID, req.Timestamp-(conversation.MsgDestructTime*1000))
+		seq, err := c.msgClient.GetLastMessageSeqByTime(ctx, conversation.ConversationID, req.Timestamp-conversation.MsgDestructTime)
 		if err != nil {
 			return nil, err
 		}

internal/rpc/msg/revoke.go

@@ -17,9 +17,8 @@ package msg
 import (
 	"context"
 	"encoding/json"
-	"time"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"time"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
@@ -80,10 +79,8 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 				switch members[req.UserID].RoleLevel {
 				case constant.GroupOwner:
 				case constant.GroupAdmin:
-					if sendMember, ok := members[msgs[0].SendID]; ok {
-						if sendMember.RoleLevel != constant.GroupOrdinaryUsers {
-							return nil, errs.ErrNoPermission.WrapMsg("no permission")
-						}
+					if members[msgs[0].SendID].RoleLevel != constant.GroupOrdinaryUsers {
+						return nil, errs.ErrNoPermission.WrapMsg("no permission")
 					}
 				default:
 					return nil, errs.ErrNoPermission.WrapMsg("no permission")

pkg/common/config/constant.go

@@ -15,10 +15,8 @@
 package config
 
 const (
-	MountConfigFilePath = "CONFIG_PATH"
-	DeploymentType      = "DEPLOYMENT_TYPE"
-	KUBERNETES          = "kubernetes"
-	ETCD                = "etcd"
+	ConfKey = "conf"
+	ETCD    = "etcd"
 )
 
 const (

pkg/common/config/env.go

@@ -1,30 +0,0 @@
-package config
-
-import "strings"
-
-var EnvPrefixMap map[string]string
-
-func init() {
-	EnvPrefixMap = make(map[string]string)
-	fileNames := []string{
-		FileName, NotificationFileName, ShareFileName, WebhooksConfigFileName,
-		KafkaConfigFileName, RedisConfigFileName,
-		MongodbConfigFileName, MinioConfigFileName, LogConfigFileName,
-		OpenIMAPICfgFileName, OpenIMCronTaskCfgFileName, OpenIMMsgGatewayCfgFileName,
-		OpenIMMsgTransferCfgFileName, OpenIMPushCfgFileName, OpenIMRPCAuthCfgFileName,
-		OpenIMRPCConversationCfgFileName, OpenIMRPCFriendCfgFileName, OpenIMRPCGroupCfgFileName,
-		OpenIMRPCMsgCfgFileName, OpenIMRPCThirdCfgFileName, OpenIMRPCUserCfgFileName, DiscoveryConfigFilename,
-	}
-
-	for _, fileName := range fileNames {
-		envKey := strings.TrimSuffix(strings.TrimSuffix(fileName, ".yml"), ".yaml")
-		envKey = "IMENV_" + envKey
-		envKey = strings.ToUpper(strings.ReplaceAll(envKey, "-", "_"))
-		EnvPrefixMap[fileName] = envKey
-	}
-}
-
-const (
-	FlagConf          = "config_folder_path"
-	FlagTransferIndex = "index"
-)

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

@@ -2,7 +2,9 @@ package redis
 
 import (
 	"context"
+	"errors"
 	"fmt"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
@@ -56,7 +58,7 @@ func callLua(ctx context.Context, rdb redis.Scripter, script *redis.Script, keys
 		}
 	}
 	v, err := r.Result()
-	if err == redis.Nil {
+	if errors.Is(err, redis.Nil) {
 		err = nil
 	}
 	return v, errs.WrapMsg(err, "call lua err", "scriptHash", script.Hash(), "keys", keys, "args", args)

pkg/common/storage/controller/auth.go

@@ -59,15 +59,16 @@ func (a *authDatabase) BatchSetTokenMapByUidPid(ctx context.Context, tokens []st
 	setMap := make(map[string]map[string]any)
 	for _, token := range tokens {
 		claims, err := tokenverify.GetClaimFromToken(token, authverify.Secret(a.accessSecret))
+		key := cachekey.GetTokenKey(claims.UserID, claims.PlatformID)
 		if err != nil {
 			continue
-		}
-		key := cachekey.GetTokenKey(claims.UserID, claims.PlatformID)
-		if v, ok := setMap[key]; ok {
-			v[token] = constant.KickedToken
 		} else {
-			setMap[key] = map[string]any{
-				token: constant.KickedToken,
+			if v, ok := setMap[key]; ok {
+				v[token] = constant.KickedToken
+			} else {
+				setMap[key] = map[string]any{
+					token: constant.KickedToken,
+				}
 			}
 		}
 	}

pkg/common/storage/controller/msg.go

@@ -18,12 +18,11 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/openimsdk/tools/utils/jsonutil"
 	"strconv"
 	"strings"
 	"time"
 
-	"github.com/openimsdk/tools/utils/jsonutil"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 
@@ -723,13 +722,13 @@ func (db *commonMsgDatabase) DeleteDoc(ctx context.Context, docID string) error
 	if index <= 0 {
 		return errs.ErrInternalServer.WrapMsg("docID is invalid", "docID", docID)
 	}
-	docIndex, err := strconv.Atoi(docID[index+1:])
+	index, err := strconv.Atoi(docID[index+1:])
 	if err != nil {
 		return errs.WrapMsg(err, "strconv.Atoi", "docID", docID)
 	}
 	conversationID := docID[:index]
 	seqs := make([]int64, db.msgTable.GetSingleGocMsgNum())
-	minSeq := db.msgTable.GetMinSeq(docIndex)
+	minSeq := db.msgTable.GetMinSeq(index)
 	for i := range seqs {
 		seqs[i] = minSeq + int64(i)
 	}

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

@@ -243,14 +243,7 @@ func (c *ConversationMgo) FindRandConversation(ctx context.Context, ts int64, li
 					"$add": []any{
 						bson.M{
 							"$toLong": "$latest_msg_destruct_time",
-						},
-						bson.M{
-							"$multiply": []any{
-								"$msg_destruct_time",
-								1000, // convert to milliseconds
-							},
-						},
-					},
+						}, "$msg_destruct_time"},
 				},
 			},
 		},

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

@@ -1091,148 +1091,22 @@ func (m *MsgMgo) onlyFindDocIndex(ctx context.Context, docID string, indexes []i
 	return msgDocModel[0].Msg, nil
 }
 
-//func (m *MsgMgo) FindSeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error) {
-//	if len(seqs) == 0 {
-//		return nil, nil
-//	}
-//	result := make([]*model.MsgInfoModel, 0, len(seqs))
-//	for docID, seqs := range m.model.GetDocIDSeqsMap(conversationID, seqs) {
-//		res, err := m.onlyFindDocIndex(ctx, docID, datautil.Slice(seqs, m.model.GetMsgIndex))
-//		if err != nil {
-//			return nil, err
-//		}
-//		for i, re := range res {
-//			if re == nil || re.Msg == nil {
-//				continue
-//			}
-//			result = append(result, res[i])
-//		}
-//	}
-//	return result, nil
-//}
-
-func (m *MsgMgo) findBeforeDocSendTime(ctx context.Context, docID string, limit int64) (int64, int64, error) {
-	if limit == 0 {
-		return 0, 0, nil
-	}
-	pipeline := []bson.M{
-		{
-			"$match": bson.M{
-				"doc_id": docID,
-			},
-		},
-		{
-			"$project": bson.M{
-				"_id":    0,
-				"doc_id": 0,
-			},
-		},
-		{
-			"$unwind": "$msgs",
-		},
-		{
-			"$project": bson.M{
-				//"_id":                0,
-				//"doc_id":             0,
-				"msgs.msg.send_time": 1,
-				"msgs.msg.seq":       1,
-			},
-		},
-	}
-	if limit > 0 {
-		pipeline = append(pipeline, bson.M{"$limit": limit})
-	}
-	type Result struct {
-		Msgs *model.MsgInfoModel `bson:"msgs"`
-	}
-	res, err := mongoutil.Aggregate[Result](ctx, m.coll, pipeline)
-	if err != nil {
-		return 0, 0, err
-	}
-	for i := len(res) - 1; i > 0; i-- {
-		v := res[i]
-		if v.Msgs != nil && v.Msgs.Msg != nil && v.Msgs.Msg.SendTime > 0 {
-			return v.Msgs.Msg.Seq, v.Msgs.Msg.SendTime, nil
-		}
-	}
-	return 0, 0, nil
-}
-
-func (m *MsgMgo) findBeforeSendTime(ctx context.Context, conversationID string, seq int64) (int64, int64, error) {
-	first := true
-	for i := m.model.GetDocIndex(seq); i >= 0; i-- {
-		limit := int64(-1)
-		if first {
-			first = false
-			limit = m.model.GetMsgIndex(seq)
-		}
-		docID := m.model.BuildDocIDByIndex(conversationID, i)
-		msgSeq, msgSendTime, err := m.findBeforeDocSendTime(ctx, docID, limit)
-		if err != nil {
-			return 0, 0, err
-		}
-		if msgSendTime > 0 {
-			return msgSeq, msgSendTime, nil
-		}
-	}
-	return 0, 0, nil
-}
-
 func (m *MsgMgo) FindSeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error) {
 	if len(seqs) == 0 {
 		return nil, nil
 	}
-	var abnormalSeq []int64
 	result := make([]*model.MsgInfoModel, 0, len(seqs))
-	for docID, docSeqs := range m.model.GetDocIDSeqsMap(conversationID, seqs) {
-		res, err := m.onlyFindDocIndex(ctx, docID, datautil.Slice(docSeqs, m.model.GetMsgIndex))
+	for docID, seqs := range m.model.GetDocIDSeqsMap(conversationID, seqs) {
+		res, err := m.onlyFindDocIndex(ctx, docID, datautil.Slice(seqs, m.model.GetMsgIndex))
 		if err != nil {
 			return nil, err
 		}
-		if len(res) == 0 {
-			abnormalSeq = append(abnormalSeq, docSeqs...)
-			continue
-		}
 		for i, re := range res {
-			if re == nil || re.Msg == nil || re.Msg.SendTime == 0 {
-				abnormalSeq = append(abnormalSeq, docSeqs[i])
+			if re == nil || re.Msg == nil {
 				continue
 			}
 			result = append(result, res[i])
 		}
 	}
-	if len(abnormalSeq) > 0 {
-		datautil.Sort(abnormalSeq, false)
-		sendTime := make(map[int64]int64)
-		var (
-			lastSeq      int64
-			lastSendTime int64
-		)
-		for _, seq := range abnormalSeq {
-			if lastSendTime > 0 && lastSeq <= seq {
-				sendTime[seq] = lastSendTime
-				continue
-			}
-			msgSeq, msgSendTime, err := m.findBeforeSendTime(ctx, conversationID, seq)
-			if err != nil {
-				return nil, err
-			}
-			if msgSendTime <= 0 {
-				break
-			}
-			sendTime[seq] = msgSendTime
-			lastSeq = msgSeq
-			lastSendTime = msgSendTime
-		}
-		for _, seq := range abnormalSeq {
-			result = append(result, &model.MsgInfoModel{
-				Msg: &model.MsgDataModel{
-					Seq:      seq,
-					Status:   constant.MsgStatusHasDeleted,
-					SendTime: sendTime[seq],
-				},
-			})
-		}
-	}
 	return result, nil
 }

pkg/common/storage/model/msg.go

@@ -15,10 +15,9 @@
 package model
 
 import (
-	"strconv"
-
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/errs"
+	"strconv"
 )
 
 const (
@@ -109,10 +108,6 @@ func (m *MsgDocModel) IsFull() bool {
 	return m.Msg[len(m.Msg)-1].Msg != nil
 }
 
-func (m *MsgDocModel) GetDocIndex(seq int64) int64 {
-	return (seq - 1) / singleGocMsgNum
-}
-
 func (m *MsgDocModel) GetDocID(conversationID string, seq int64) string {
 	seqSuffix := (seq - 1) / singleGocMsgNum
 	return m.indexGen(conversationID, seqSuffix)
@@ -140,10 +135,6 @@ func (*MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
 	return conversationID + ":" + strconv.FormatInt(seqSuffix, 10)
 }
 
-func (*MsgDocModel) BuildDocIDByIndex(conversationID string, index int64) string {
-	return conversationID + ":" + strconv.FormatInt(index, 10)
-}
-
 func (*MsgDocModel) GenExceptionMessageBySeqs(seqs []int64) (exceptionMsg []*sdkws.MsgData) {
 	for _, v := range seqs {
 		msgModel := new(sdkws.MsgData)

tools/seq/internal/main.go

@@ -5,6 +5,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
+	"github.com/openimsdk/tools/db/mongoutil"
+	"github.com/openimsdk/tools/db/redisutil"
+	"github.com/redis/go-redis/v9"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"gopkg.in/yaml.v3"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -14,19 +24,6 @@ import (
 	"sync/atomic"
 	"syscall"
 	"time"
-
-	"github.com/mitchellh/mapstructure"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
-	"github.com/openimsdk/tools/db/mongoutil"
-	"github.com/openimsdk/tools/db/redisutil"
-	"github.com/openimsdk/tools/utils/runtimeenv"
-	"github.com/redis/go-redis/v9"
-	"github.com/spf13/viper"
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
 const (
@@ -44,22 +41,12 @@ const (
 )
 
 func readConfig[T any](dir string, name string) (*T, error) {
-	if runtimeenv.PrintRuntimeEnvironment() == config.KUBERNETES {
-		dir = os.Getenv(config.MountConfigFilePath)
-	}
-	v := viper.New()
-	v.SetEnvPrefix(config.EnvPrefixMap[name])
-	v.AutomaticEnv()
-	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
-	v.SetConfigFile(filepath.Join(dir, name))
-	if err := v.ReadInConfig(); err != nil {
+	data, err := os.ReadFile(filepath.Join(dir, name))
+	if err != nil {
 		return nil, err
 	}
-	fn := func(config *mapstructure.DecoderConfig) {
-		config.TagName = "mapstructure"
-	}
 	var conf T
-	if err := v.Unmarshal(&conf, fn); err != nil {
+	if err := yaml.Unmarshal(data, &conf); err != nil {
 		return nil, err
 	}
 	return &conf, nil

tools/seq/main.go

@@ -3,10 +3,8 @@ package main
 import (
 	"flag"
 	"fmt"
-	"os"
-	"time"
-
 	"github.com/openimsdk/open-im-server/v3/tools/seq/internal"
+	"time"
 )
 
 func main() {
@@ -19,8 +17,6 @@ func main() {
 	flag.Parse()
 	if err := internal.Main(config, time.Duration(second)*time.Second); err != nil {
 		fmt.Println("seq task", err)
-		os.Exit(1)
-		return
 	}
 	fmt.Println("seq task success!")
 }

version/version

@@ -1 +1 @@
-v3.8.3-patch.1
+v3.8.3