Merge branch 'main' of github.com:OpenIMSDK/Open-IM-Server

internal/push/push_handler.go

@@ -65,7 +65,13 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 	case constant.SuperGroupChatType:
 		err = c.pusher.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
 	default:
-		err = c.pusher.Push2User(ctx, []string{pbData.MsgData.SendID, pbData.MsgData.RecvID}, pbData.MsgData)
+		var pushUserIDs []string
+		if pbData.MsgData.SendID != pbData.MsgData.RecvID {
+			pushUserIDs = []string{pbData.MsgData.SendID, pbData.MsgData.RecvID}
+		} else {
+			pushUserIDs = []string{pbData.MsgData.SendID}
+		}
+		err = c.pusher.Push2User(ctx, pushUserIDs, pbData.MsgData)
 	}
 	if err != nil {
 		if err == errNoOfflinePusher {

internal/rpc/msg/server.go

@@ -42,7 +42,6 @@ type (
 		friend                 *rpcclient.FriendRpcClient
 		GroupLocalCache        *localcache.GroupLocalCache
 		ConversationLocalCache *localcache.ConversationLocalCache
-		MessageLocker          MessageLocker
 		Handlers               MessageInterceptorChain
 		notificationSender     *rpcclient.NotificationSender
 	}
@@ -91,7 +90,6 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
 		GroupLocalCache:        localcache.NewGroupLocalCache(&groupRpcClient),
 		ConversationLocalCache: localcache.NewConversationLocalCache(&conversationClient),
 		friend:                 &friendRpcClient,
-		MessageLocker:          NewLockerMessage(cacheModel),
 	}
 	s.notificationSender = rpcclient.NewNotificationSender(rpcclient.WithLocalSendMsg(s.SendMsg))
 	s.addInterceptorHandler(MessageHasReadEnabled)

internal/rpc/msg/sync_msg.go

@@ -95,6 +95,7 @@ func (m *msgServer) GetMaxSeq(ctx context.Context, req *sdkws.GetMaxSeqReq) (*sd
 	for _, conversationID := range conversationIDs {
 		conversationIDs = append(conversationIDs, utils.GetNotificationConversationIDByConversationID(conversationID))
 	}
+	conversationIDs = append(conversationIDs, utils.GetSelfNotificationConversationID(req.UserID))
 	log.ZDebug(ctx, "GetMaxSeq", "conversationIDs", conversationIDs)
 	maxSeqs, err := m.MsgDatabase.GetMaxSeqs(ctx, conversationIDs)
 	if err != nil {

internal/rpc/user/user.go

@@ -67,7 +67,7 @@ func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
 		return errors.New("len(config.Config.Manager.AppManagerUid) != len(config.Config.Manager.Nickname)")
 	}
 	for k, v := range config.Config.Manager.UserID {
-		users = append(users, &tablerelation.UserModel{UserID: v, Nickname: config.Config.Manager.Nickname[k]})
+		users = append(users, &tablerelation.UserModel{UserID: v, Nickname: config.Config.Manager.Nickname[k], AppMangerLevel: constant.AppAdmin})
 	}
 	userDB := relation.NewUserGorm(db)
 	cache := cache.NewUserCacheRedis(rdb, userDB, cache.GetDefaultOpt())

internal/tools/conversation.go

@@ -33,6 +33,7 @@ func (c *MsgTool) ConversationsDestructMsgs() {
 	}
 	log.ZDebug(context.Background(), "nums conversations need destruct", "nums", len(conversations))
 	for _, conversation := range conversations {
+		ctx = mcontext.NewCtx(utils.GetSelfFuncName() + "-" + utils.OperationIDGenerator() + "-" + conversation.ConversationID + "-" + conversation.OwnerUserID)
 		log.ZDebug(
 			ctx,
 			"UserMsgsDestruct",
@@ -45,16 +46,17 @@ func (c *MsgTool) ConversationsDestructMsgs() {
 			"lastMsgDestructTime",
 			conversation.LatestMsgDestructTime,
 		)
+		now := time.Now()
 		seqs, err := c.msgDatabase.UserMsgsDestruct(ctx, conversation.OwnerUserID, conversation.ConversationID, conversation.MsgDestructTime, conversation.LatestMsgDestructTime)
 		if err != nil {
 			log.ZError(ctx, "user msg destruct failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
 			continue
 		}
-		if err := c.conversationDatabase.UpdateUsersConversationFiled(ctx, []string{conversation.OwnerUserID}, conversation.ConversationID, map[string]interface{}{"latest_msg_destruct_time": time.Now()}); err != nil {
-			log.ZError(ctx, "updateUsersConversationFiled failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
-			continue
-		}
 		if len(seqs) > 0 {
+			if err := c.conversationDatabase.UpdateUsersConversationFiled(ctx, []string{conversation.OwnerUserID}, conversation.ConversationID, map[string]interface{}{"latest_msg_destruct_time": now}); err != nil {
+				log.ZError(ctx, "updateUsersConversationFiled failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
+				continue
+			}
 			if err := c.msgNotificationSender.UserDeleteMsgsNotification(ctx, conversation.OwnerUserID, conversation.ConversationID, seqs); err != nil {
 				log.ZError(ctx, "userDeleteMsgsNotification failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
 			}

internal/tools/cron_task.go

@@ -41,11 +41,11 @@ func StartCronTask() error {
 		panic(err)
 	}
 	log.ZInfo(context.Background(), "start msgDestruct cron task", "cron config", config.Config.MsgDestructTime)
-	// _, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs)
+	_, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs)
-	// if err != nil {
+	if err != nil {
-	// 	fmt.Println("start conversationsDestructMsgs cron failed", err.Error(), config.Config.ChatRecordsClearTime)
+		fmt.Println("start conversationsDestructMsgs cron failed", err.Error(), config.Config.ChatRecordsClearTime)
-	// 	panic(err)
+		panic(err)
-	// }
+	}
 	c.Start()
 	wg.Wait()
 	return nil

pkg/common/db/controller/msg.go

@@ -63,7 +63,7 @@ type CommonMsgDatabase interface {
 	// 删除会话消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache)
 	DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
 	// 用户标记删除过期消息返回标记删除的seq列表
-	UserMsgsDestruct(cte context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error)
+	UserMsgsDestruct(ctx context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error)
 
 	// 用户根据seq删除消息
 	DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error
@@ -641,7 +641,7 @@ func (db *commonMsgDatabase) UserMsgsDestruct(ctx context.Context, userID string
 		if err != nil || msgDocModel.DocID == "" {
 			if err != nil {
 				if err == unrelation.ErrMsgListNotExist {
-					log.ZDebug(ctx, "deleteMsgRecursion finished", "conversationID", conversationID, "userID", userID, "index", index)
+					log.ZDebug(ctx, "not doc find", "conversationID", conversationID, "userID", userID, "index", index)
 				} else {
 					log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
 				}
@@ -652,25 +652,38 @@ func (db *commonMsgDatabase) UserMsgsDestruct(ctx context.Context, userID string
 		index++
 		//&& msgDocModel.Msg[0].Msg.SendTime > lastMsgDestructTime.UnixMilli()
 		if len(msgDocModel.Msg) > 0 {
+			i := 0
+			var over bool
 			for _, msg := range msgDocModel.Msg {
+				i++
 				if msg != nil && msg.Msg != nil && msg.Msg.SendTime+destructTime*1000 <= time.Now().UnixMilli() {
-					if msg.Msg.SendTime > lastMsgDestructTime.UnixMilli() && !utils.Contain(userID, msg.DelList...) {
+					if msg.Msg.SendTime+destructTime*1000 > lastMsgDestructTime.UnixMilli() && !utils.Contain(userID, msg.DelList...) {
 						seqs = append(seqs, msg.Msg.Seq)
 					}
 				} else {
-					log.ZDebug(ctx, "deleteMsgRecursion finished", "conversationID", conversationID, "userID", userID, "index", index)
+					log.ZDebug(ctx, "all msg need destruct is found", "conversationID", conversationID, "userID", userID, "index", index, "stop index", i)
+					over = true
 					break
 				}
 			}
+			if over {
+				break
+			}
 		}
 	}
 
 	log.ZDebug(ctx, "UserMsgsDestruct", "conversationID", conversationID, "userID", userID, "seqs", seqs)
 	if len(seqs) > 0 {
-		latestSeq := seqs[len(seqs)-1]
+		userMinSeq := seqs[len(seqs)-1] + 1
-		if err := db.cache.SetConversationUserMinSeq(ctx, conversationID, userID, latestSeq); err != nil {
+		currentUserMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
+		if err != nil && errs.Unwrap(err) != redis.Nil {
 			return nil, err
 		}
+		if currentUserMinSeq < userMinSeq {
+			if err := db.cache.SetConversationUserMinSeq(ctx, conversationID, userID, userMinSeq); err != nil {
+				return nil, err
+			}
+		}
 	}
 	return seqs, nil
 }

pkg/common/db/relation/conversation_model.go

@@ -207,7 +207,7 @@ func (c *ConversationGorm) GetConversationIDsNeedDestruct(
 ) (conversations []*relation.ConversationModel, err error) {
 	return conversations, utils.Wrap(
 		c.db(ctx).
-			Where("is_msg_destruct = 1 && UNIX_TIMESTAMP(NOW()) > (msg_destruct_time + UNIX_TIMESTAMP(latest_msg_destruct_time)) && msg_destruct_time != 0").
+			Where("is_msg_destruct = 1 && msg_destruct_time != 0 && (UNIX_TIMESTAMP(NOW()) > (msg_destruct_time + UNIX_TIMESTAMP(latest_msg_destruct_time)) || latest_msg_destruct_time is NULL)").
 			Find(&conversations).
 			Error,
 		"",

pkg/common/db/s3/minio/minio.go

@@ -22,6 +22,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/minio/minio-go/v7"
@@ -56,21 +57,20 @@ func NewMinio() (s3.Interface, error) {
 	if err != nil {
 		return nil, err
 	}
-	exists, err := client.BucketExists(context.Background(), conf.Bucket)
+	m := &Minio{
-	if err != nil {
-		return nil, err
-	}
-	if !exists {
-		if err := client.MakeBucket(context.Background(), conf.Bucket, minio.MakeBucketOptions{}); err != nil {
-			return nil, err
-		}
-	}
-	return &Minio{
 		bucket:    conf.Bucket,
 		bucketURL: conf.Endpoint + "/" + conf.Bucket + "/",
 		opts:      opts,
 		core:      &minio.Core{Client: client},
-	}, nil
+		lock:      &sync.Mutex{},
+		init:      false,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := m.initMinio(ctx); err != nil {
+		fmt.Println("init minio error:", err)
+	}
+	return m, nil
 }
 
 type Minio struct {
@@ -78,6 +78,30 @@ type Minio struct {
 	bucketURL string
 	opts      *minio.Options
 	core      *minio.Core
+	lock      sync.Locker
+	init      bool
+}
+
+func (m *Minio) initMinio(ctx context.Context) error {
+	if m.init {
+		return nil
+	}
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if m.init {
+		return nil
+	}
+	exists, err := m.core.Client.BucketExists(ctx, config.Config.Object.Minio.Bucket)
+	if err != nil {
+		return fmt.Errorf("check bucket exists error: %w", err)
+	}
+	if !exists {
+		if err := m.core.Client.MakeBucket(ctx, config.Config.Object.Minio.Bucket, minio.MakeBucketOptions{}); err != nil {
+			return fmt.Errorf("make bucket error: %w", err)
+		}
+	}
+	m.init = true
+	return nil
 }
 
 func (m *Minio) Engine() string {
@@ -93,6 +117,9 @@ func (m *Minio) PartLimit() *s3.PartLimit {
 }
 
 func (m *Minio) InitiateMultipartUpload(ctx context.Context, name string) (*s3.InitiateMultipartUploadResult, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return nil, err
+	}
 	uploadID, err := m.core.NewMultipartUpload(ctx, m.bucket, name, minio.PutObjectOptions{})
 	if err != nil {
 		return nil, err
@@ -105,6 +132,9 @@ func (m *Minio) InitiateMultipartUpload(ctx context.Context, name string) (*s3.I
 }
 
 func (m *Minio) CompleteMultipartUpload(ctx context.Context, uploadID string, name string, parts []s3.Part) (*s3.CompleteMultipartUploadResult, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return nil, err
+	}
 	minioParts := make([]minio.CompletePart, len(parts))
 	for i, part := range parts {
 		minioParts[i] = minio.CompletePart{
@@ -142,6 +172,9 @@ func (m *Minio) PartSize(ctx context.Context, size int64) (int64, error) {
 }
 
 func (m *Minio) AuthSign(ctx context.Context, uploadID string, name string, expire time.Duration, partNumbers []int) (*s3.AuthSignResult, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return nil, err
+	}
 	creds, err := m.opts.Creds.Get()
 	if err != nil {
 		return nil, err
@@ -170,6 +203,9 @@ func (m *Minio) AuthSign(ctx context.Context, uploadID string, name string, expi
 }
 
 func (m *Minio) PresignedPutObject(ctx context.Context, name string, expire time.Duration) (string, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return "", err
+	}
 	rawURL, err := m.core.Client.PresignedPutObject(ctx, m.bucket, name, expire)
 	if err != nil {
 		return "", err
@@ -178,10 +214,16 @@ func (m *Minio) PresignedPutObject(ctx context.Context, name string, expire time
 }
 
 func (m *Minio) DeleteObject(ctx context.Context, name string) error {
+	if err := m.initMinio(ctx); err != nil {
+		return err
+	}
 	return m.core.Client.RemoveObject(ctx, m.bucket, name, minio.RemoveObjectOptions{})
 }
 
 func (m *Minio) StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return nil, err
+	}
 	info, err := m.core.Client.StatObject(ctx, m.bucket, name, minio.StatObjectOptions{})
 	if err != nil {
 		return nil, err
@@ -195,6 +237,9 @@ func (m *Minio) StatObject(ctx context.Context, name string) (*s3.ObjectInfo, er
 }
 
 func (m *Minio) CopyObject(ctx context.Context, src string, dst string) (*s3.CopyObjectInfo, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return nil, err
+	}
 	result, err := m.core.Client.CopyObject(ctx, minio.CopyDestOptions{
 		Bucket: m.bucket,
 		Object: dst,
@@ -226,10 +271,16 @@ func (m *Minio) IsNotFound(err error) bool {
 }
 
 func (m *Minio) AbortMultipartUpload(ctx context.Context, uploadID string, name string) error {
+	if err := m.initMinio(ctx); err != nil {
+		return err
+	}
 	return m.core.AbortMultipartUpload(ctx, m.bucket, name, uploadID)
 }
 
 func (m *Minio) ListUploadedParts(ctx context.Context, uploadID string, name string, partNumberMarker int, maxParts int) (*s3.ListUploadedPartsResult, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return nil, err
+	}
 	result, err := m.core.ListObjectParts(ctx, m.bucket, name, uploadID, partNumberMarker, maxParts)
 	if err != nil {
 		return nil, err
@@ -253,6 +304,9 @@ func (m *Minio) ListUploadedParts(ctx context.Context, uploadID string, name str
 }
 
 func (m *Minio) AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (string, error) {
+	if err := m.initMinio(ctx); err != nil {
+		return "", err
+	}
 	reqParams := make(url.Values)
 	if opt != nil {
 		if opt.ContentType != "" {

pkg/common/db/table/relation/user.go

@@ -29,7 +29,7 @@ type UserModel struct {
 	FaceURL          string    `gorm:"column:face_url;size:255"`
 	Ex               string    `gorm:"column:ex;size:1024"`
 	CreateTime       time.Time `gorm:"column:create_time;index:create_time;autoCreateTime"`
-	AppMangerLevel   int32     `gorm:"column:app_manger_level;default:18"`
+	AppMangerLevel   int32     `gorm:"column:app_manger_level;default:1"`
 	GlobalRecvMsgOpt int32     `gorm:"column:global_recv_msg_opt"`
 }
 

pkg/common/locker/message_locker.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package msg
+package locker
 
 import (
 	"context"

pkg/rpcclient/msg.go

@@ -71,8 +71,9 @@ func newContentTypeConf() map[int32]config.NotificationConf {
 		constant.ConversationUnreadNotification:      config.Config.Notification.ConversationChanged,
 		constant.ConversationPrivateChatNotification: config.Config.Notification.ConversationSetPrivate,
 		// msg
-		constant.MsgRevokeNotification: {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.MsgRevokeNotification:  {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
-		constant.HasReadReceipt:        {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.HasReadReceipt:         {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
+		constant.DeleteMsgsNotification: {IsSendMsg: false, ReliabilityLevel: constant.ReliableNotificationNoMsg},
 	}
 }
 
@@ -116,7 +117,7 @@ func newSessionTypeConf() map[int32]int32 {
 		constant.ConversationUnreadNotification:      constant.SingleChatType,
 		constant.ConversationPrivateChatNotification: constant.SingleChatType,
 		// delete
-		constant.MsgDeleteNotification: constant.SingleChatType,
+		constant.DeleteMsgsNotification: constant.SingleChatType,
 	}
 }
 

pkg/rpcclient/notification/msg.go

@@ -36,7 +36,7 @@ func (m *MsgNotificationSender) UserDeleteMsgsNotification(ctx context.Context,
 		ConversationID: conversationID,
 		Seqs:           seqs,
 	}
-	return m.Notification(ctx, userID, userID, constant.MsgDeleteNotification, &tips)
+	return m.Notification(ctx, userID, userID, constant.DeleteMsgsNotification, &tips)
 }
 
 func (m *MsgNotificationSender) MarkAsReadNotification(ctx context.Context, conversationID string, sesstionType int32, sendID, recvID string, seqs []int64, hasReadSeq int64) error {

pkg/utils/utils.go

@@ -367,6 +367,10 @@ func GetNotificationConversationIDByConversationID(conversationID string) string
 	return ""
 }
 
+func GetSelfNotificationConversationID(userID string) string {
+	return "n_" + userID + "_" + userID
+}
+
 func GetSeqsBeginEnd(seqs []int64) (int64, int64) {
 	if len(seqs) == 0 {
 		return 0, 0