mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-06-25 20:00:27 +08:00
pin group message
This commit is contained in:
parent
6e42d00794
commit
d51378bd7b
@ -173,3 +173,15 @@ func (o *GroupApi) GetGroupApplicationUnhandledCount(c *gin.Context) {
|
||||
func (o *GroupApi) GetCommonGroupsWithFriend(c *gin.Context) {
|
||||
a2r.Call(c, group.GroupClient.GetCommonGroupsWithFriend, o.Client)
|
||||
}
|
||||
|
||||
func (o *GroupApi) PinGroupMessage(c *gin.Context) {
|
||||
a2r.Call(c, group.GroupClient.PinGroupMessage, o.Client)
|
||||
}
|
||||
|
||||
func (o *GroupApi) UnpinGroupMessage(c *gin.Context) {
|
||||
a2r.Call(c, group.GroupClient.UnpinGroupMessage, o.Client)
|
||||
}
|
||||
|
||||
func (o *GroupApi) GetGroupPinnedMessages(c *gin.Context) {
|
||||
a2r.Call(c, group.GroupClient.GetGroupPinnedMessages, o.Client)
|
||||
}
|
||||
|
||||
@ -243,6 +243,9 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
|
||||
groupRouterGroup.POST("/get_full_join_group_ids", g.GetFullJoinGroupIDs)
|
||||
groupRouterGroup.POST("/get_group_application_unhandled_count", g.GetGroupApplicationUnhandledCount)
|
||||
groupRouterGroup.POST("/get_common_groups_with_friend", g.GetCommonGroupsWithFriend)
|
||||
groupRouterGroup.POST("/pin_group_message", g.PinGroupMessage)
|
||||
groupRouterGroup.POST("/unpin_group_message", g.UnpinGroupMessage)
|
||||
groupRouterGroup.POST("/get_group_pinned_messages", g.GetGroupPinnedMessages)
|
||||
}
|
||||
// certificate
|
||||
{
|
||||
|
||||
@ -101,6 +101,10 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
groupPinnedMsgDB, err := mgo.NewGroupPinnedMsgMongo(mgocli.GetDB())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
//userRpcClient := rpcclient.NewUserRpcClient(client, config.Share.RpcRegisterName.User, config.Share.IMAdminUserID)
|
||||
//msgRpcClient := rpcclient.NewMessageRpcClient(client, config.Share.RpcRegisterName.Msg)
|
||||
@ -130,7 +134,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
|
||||
conversationClient: rpcli.NewConversationClient(conversationConn),
|
||||
//cryptoClient: rpcli.NewCryptoClient(cryptoConn),
|
||||
}
|
||||
gs.db = controller.NewGroupDatabase(rdb, &config.LocalCacheConfig, groupDB, groupMemberDB, groupRequestDB, mgocli.GetTx(), grouphash.NewGroupHashFromGroupServer(&gs))
|
||||
gs.db = controller.NewGroupDatabase(rdb, &config.LocalCacheConfig, groupDB, groupMemberDB, groupRequestDB, groupPinnedMsgDB, mgocli.GetTx(), grouphash.NewGroupHashFromGroupServer(&gs))
|
||||
gs.notification = NewNotificationSender(gs.db, config, gs.userClient, gs.msgClient, gs.conversationClient)
|
||||
localcache.InitLocalCache(&config.LocalCacheConfig)
|
||||
pbgroup.RegisterGroupServer(server, &gs)
|
||||
|
||||
@ -852,6 +852,32 @@ func (g *NotificationSender) GroupMemberSetToAdminNotification(ctx context.Conte
|
||||
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToAdminNotification, tips)
|
||||
}
|
||||
|
||||
// GroupMessagePinnedNotification 通知群成员有消息被置顶或取消置顶
|
||||
// pinType: 1=置顶, 2=取消置顶
|
||||
func (g *NotificationSender) GroupMessagePinnedNotification(ctx context.Context, groupID string, pinType int32,
|
||||
pinned *sdkws.GroupPinnedMsgInfo, pinnedList []*sdkws.GroupPinnedMsgInfo) {
|
||||
var err error
|
||||
defer func() {
|
||||
if err != nil {
|
||||
log.ZError(ctx, stringutil.GetFuncName(1)+" failed", err)
|
||||
}
|
||||
}()
|
||||
groupInfo, err := g.getGroupInfo(ctx, groupID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
tips := &sdkws.GroupMessagePinnedTips{
|
||||
Group: groupInfo,
|
||||
Type: pinType,
|
||||
PinnedMsg: pinned,
|
||||
PinnedList: pinnedList,
|
||||
}
|
||||
if err = g.fillOpUser(ctx, &tips.OpUser, groupID); err != nil {
|
||||
return
|
||||
}
|
||||
g.Notification(ctx, mcontext.GetOpUserID(ctx), groupID, constant.GroupMessagePinnedNotification, tips)
|
||||
}
|
||||
|
||||
func (g *NotificationSender) GroupMemberSetToOrdinaryUserNotification(ctx context.Context, groupID, groupMemberUserID string) {
|
||||
var err error
|
||||
defer func() {
|
||||
|
||||
271
internal/rpc/group/pinned_msg.go
Normal file
271
internal/rpc/group/pinned_msg.go
Normal file
@ -0,0 +1,271 @@
|
||||
// Copyright © 2026 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.
|
||||
|
||||
package group
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
pbgroup "github.com/openimsdk/protocol/group"
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/mcontext"
|
||||
)
|
||||
|
||||
// 群置顶消息相关 RPC 实现:
|
||||
// - 自动滚动保留最近 N 条置顶消息(N=model.GroupPinnedMsgMaxKeep,默认为 3)
|
||||
// - 置顶时把整条消息内容做完整快照存档,避免后续消息删除/撤回影响展示
|
||||
// - 每条置顶记录拥有唯一 pinID,作为 unpin 时的精准删除凭据
|
||||
// - 权限:默认全员可置顶;当 group.AllowPinMsg=1 时,仅群主/管理员可置顶或取消置顶
|
||||
|
||||
const (
|
||||
groupPinnedActionPin = int32(1)
|
||||
groupPinnedActionUnpin = int32(2)
|
||||
)
|
||||
|
||||
// PinGroupMessage 群聊中置顶单条消息
|
||||
func (s *groupServer) PinGroupMessage(ctx context.Context, req *pbgroup.PinGroupMessageReq) (*pbgroup.PinGroupMessageResp, error) {
|
||||
if req.GroupID == "" {
|
||||
return nil, errs.ErrArgs.WrapMsg("groupID empty")
|
||||
}
|
||||
if req.Seq <= 0 {
|
||||
return nil, errs.ErrArgs.WrapMsg("seq must be positive")
|
||||
}
|
||||
|
||||
group, err := s.db.TakeGroup(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if group.Status == constant.GroupStatusDismissed {
|
||||
return nil, servererrs.ErrDismissedAlready.Wrap()
|
||||
}
|
||||
|
||||
if err := s.checkPinPermission(ctx, group); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
|
||||
msgData, err := s.msgClient.GetSingleMsgBySeq(ctx, conversationID, req.Seq)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if msgData == nil {
|
||||
return nil, servererrs.ErrRecordNotFound.WrapMsg("message not found by seq")
|
||||
}
|
||||
if msgData.GroupID != "" && msgData.GroupID != req.GroupID {
|
||||
return nil, errs.ErrArgs.WrapMsg("seq does not belong to this group")
|
||||
}
|
||||
if msgData.Status >= constant.MsgStatusHasDeleted {
|
||||
return nil, servererrs.ErrRecordNotFound.WrapMsg("message has been deleted")
|
||||
}
|
||||
|
||||
pin := buildPinSnapshot(req.GroupID, conversationID, mcontext.GetOpUserID(ctx), msgData)
|
||||
|
||||
pinnedList, err := s.db.PinGroupMessage(ctx, req.GroupID, pin)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pbPinned := pinnedMsgDB2PB(pin)
|
||||
pbList := pinnedListDB2PB(pinnedList)
|
||||
|
||||
s.notification.GroupMessagePinnedNotification(ctx, req.GroupID, groupPinnedActionPin, pbPinned, pbList)
|
||||
|
||||
return &pbgroup.PinGroupMessageResp{
|
||||
PinnedMsg: pbPinned,
|
||||
PinnedList: pbList,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// UnpinGroupMessage 群聊中取消置顶单条消息(pinID 优先;为空则按 seq)
|
||||
func (s *groupServer) UnpinGroupMessage(ctx context.Context, req *pbgroup.UnpinGroupMessageReq) (*pbgroup.UnpinGroupMessageResp, error) {
|
||||
if req.GroupID == "" {
|
||||
return nil, errs.ErrArgs.WrapMsg("groupID empty")
|
||||
}
|
||||
if req.PinID == "" && req.Seq <= 0 {
|
||||
return nil, errs.ErrArgs.WrapMsg("either pinID or seq must be provided")
|
||||
}
|
||||
|
||||
group, err := s.db.TakeGroup(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if group.Status == constant.GroupStatusDismissed {
|
||||
return nil, servererrs.ErrDismissedAlready.Wrap()
|
||||
}
|
||||
if err := s.checkPinPermission(ctx, group); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
current, err := s.db.GetGroupPinnedMessages(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var target *model.GroupPinnedMessage
|
||||
for _, m := range current {
|
||||
if req.PinID != "" {
|
||||
if m.PinID == req.PinID {
|
||||
target = m
|
||||
break
|
||||
}
|
||||
} else if m.Seq == req.Seq {
|
||||
target = m
|
||||
break
|
||||
}
|
||||
}
|
||||
if target == nil {
|
||||
return nil, servererrs.ErrRecordNotFound.WrapMsg("pinned message not found")
|
||||
}
|
||||
|
||||
pinnedList, err := s.db.UnpinGroupMessage(ctx, req.GroupID, req.PinID, req.Seq)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pbPinned := pinnedMsgDB2PB(target)
|
||||
pbList := pinnedListDB2PB(pinnedList)
|
||||
|
||||
s.notification.GroupMessagePinnedNotification(ctx, req.GroupID, groupPinnedActionUnpin, pbPinned, pbList)
|
||||
|
||||
return &pbgroup.UnpinGroupMessageResp{PinnedList: pbList}, nil
|
||||
}
|
||||
|
||||
// GetGroupPinnedMessages 获取群置顶消息列表
|
||||
func (s *groupServer) GetGroupPinnedMessages(ctx context.Context, req *pbgroup.GetGroupPinnedMessagesReq) (*pbgroup.GetGroupPinnedMessagesResp, error) {
|
||||
if req.GroupID == "" {
|
||||
return nil, errs.ErrArgs.WrapMsg("groupID empty")
|
||||
}
|
||||
if err := s.checkAdminOrInGroup(ctx, req.GroupID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pinnedList, err := s.db.GetGroupPinnedMessages(ctx, req.GroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbgroup.GetGroupPinnedMessagesResp{
|
||||
PinnedList: pinnedListDB2PB(pinnedList),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// checkPinPermission 校验当前操作者是否具备群消息置顶权限
|
||||
func (s *groupServer) checkPinPermission(ctx context.Context, group *model.Group) error {
|
||||
if authverify.IsAppManagerUid(ctx, s.config.Share.IMAdminUserID) {
|
||||
return nil
|
||||
}
|
||||
opUserID := mcontext.GetOpUserID(ctx)
|
||||
if opUserID == "" {
|
||||
return errs.ErrNoPermission.WrapMsg("op user id empty")
|
||||
}
|
||||
member, err := s.db.TakeGroupMember(ctx, group.GroupID, opUserID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
isOwnerOrAdmin := member.RoleLevel == constant.GroupOwner || member.RoleLevel == constant.GroupAdmin
|
||||
if group.AllowPinMsg == model.GroupPermAdminOnly && !isOwnerOrAdmin {
|
||||
return errs.ErrNoPermission.WrapMsg("only owner or admin can pin/unpin group message")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// buildPinSnapshot 把 sdkws.MsgData 完整快照成 GroupPinnedMessage
|
||||
// PinID 在 mgo 层 Pin 时若为空会自动生成;这里留空交由存储层处理
|
||||
func buildPinSnapshot(groupID, conversationID, opUserID string, m *sdkws.MsgData) *model.GroupPinnedMessage {
|
||||
pin := &model.GroupPinnedMessage{
|
||||
GroupID: groupID,
|
||||
ConversationID: conversationID,
|
||||
Seq: m.Seq,
|
||||
ServerMsgID: m.ServerMsgID,
|
||||
ClientMsgID: m.ClientMsgID,
|
||||
SendID: m.SendID,
|
||||
RecvID: m.RecvID,
|
||||
SenderPlatformID: m.SenderPlatformID,
|
||||
SenderNickname: m.SenderNickname,
|
||||
SenderFaceURL: m.SenderFaceURL,
|
||||
SessionType: m.SessionType,
|
||||
MsgFrom: m.MsgFrom,
|
||||
ContentType: m.ContentType,
|
||||
Content: string(m.Content),
|
||||
AtUserIDList: append([]string(nil), m.AtUserIDList...),
|
||||
Options: copyOptions(m.Options),
|
||||
AttachedInfo: m.AttachedInfo,
|
||||
Ex: m.Ex,
|
||||
SendTime: m.SendTime,
|
||||
CreateTime: m.CreateTime,
|
||||
Status: m.Status,
|
||||
PinUserID: opUserID,
|
||||
PinTime: time.Now().UnixMilli(),
|
||||
}
|
||||
if m.OfflinePushInfo != nil {
|
||||
pin.OfflinePush = &model.GroupPinnedOfflinePush{
|
||||
Title: m.OfflinePushInfo.Title,
|
||||
Desc: m.OfflinePushInfo.Desc,
|
||||
Ex: m.OfflinePushInfo.Ex,
|
||||
IOSPushSound: m.OfflinePushInfo.IOSPushSound,
|
||||
IOSBadgeCount: m.OfflinePushInfo.IOSBadgeCount,
|
||||
SignalInfo: m.OfflinePushInfo.SignalInfo,
|
||||
}
|
||||
}
|
||||
return pin
|
||||
}
|
||||
|
||||
func copyOptions(src map[string]bool) map[string]bool {
|
||||
if len(src) == 0 {
|
||||
return nil
|
||||
}
|
||||
dst := make(map[string]bool, len(src))
|
||||
for k, v := range src {
|
||||
dst[k] = v
|
||||
}
|
||||
return dst
|
||||
}
|
||||
|
||||
func pinnedMsgDB2PB(m *model.GroupPinnedMessage) *sdkws.GroupPinnedMsgInfo {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
return &sdkws.GroupPinnedMsgInfo{
|
||||
PinID: m.PinID,
|
||||
GroupID: m.GroupID,
|
||||
ConversationID: m.ConversationID,
|
||||
Seq: m.Seq,
|
||||
ServerMsgID: m.ServerMsgID,
|
||||
ClientMsgID: m.ClientMsgID,
|
||||
SendID: m.SendID,
|
||||
RecvID: m.RecvID,
|
||||
SenderPlatformID: m.SenderPlatformID,
|
||||
SenderNickname: m.SenderNickname,
|
||||
SenderFaceURL: m.SenderFaceURL,
|
||||
SessionType: m.SessionType,
|
||||
MsgFrom: m.MsgFrom,
|
||||
ContentType: m.ContentType,
|
||||
Content: m.Content,
|
||||
AtUserIDList: append([]string(nil), m.AtUserIDList...),
|
||||
Options: copyOptions(m.Options),
|
||||
AttachedInfo: m.AttachedInfo,
|
||||
Ex: m.Ex,
|
||||
SendTime: m.SendTime,
|
||||
CreateTime: m.CreateTime,
|
||||
Status: m.Status,
|
||||
PinUserID: m.PinUserID,
|
||||
PinTime: m.PinTime,
|
||||
}
|
||||
}
|
||||
|
||||
func pinnedListDB2PB(list []*model.GroupPinnedMessage) []*sdkws.GroupPinnedMsgInfo {
|
||||
if len(list) == 0 {
|
||||
return nil
|
||||
}
|
||||
result := make([]*sdkws.GroupPinnedMsgInfo, 0, len(list))
|
||||
for _, m := range list {
|
||||
result = append(result, pinnedMsgDB2PB(m))
|
||||
}
|
||||
return result
|
||||
}
|
||||
@ -126,6 +126,11 @@ type GroupDatabase interface {
|
||||
FindJoinGroupID(ctx context.Context, userID string) ([]string, error)
|
||||
|
||||
GetGroupApplicationUnhandledCount(ctx context.Context, groupIDs []string, ts int64) (int64, error)
|
||||
|
||||
// 群置顶消息:保留最近 N 条
|
||||
PinGroupMessage(ctx context.Context, groupID string, msg *model.GroupPinnedMessage) ([]*model.GroupPinnedMessage, error)
|
||||
UnpinGroupMessage(ctx context.Context, groupID string, pinID string, seq int64) ([]*model.GroupPinnedMessage, error)
|
||||
GetGroupPinnedMessages(ctx context.Context, groupID string) ([]*model.GroupPinnedMessage, error)
|
||||
}
|
||||
|
||||
func NewGroupDatabase(
|
||||
@ -134,24 +139,39 @@ func NewGroupDatabase(
|
||||
groupDB database.Group,
|
||||
groupMemberDB database.GroupMember,
|
||||
groupRequestDB database.GroupRequest,
|
||||
groupPinnedMsgDB database.GroupPinnedMsg,
|
||||
ctxTx tx.Tx,
|
||||
groupHash cache.GroupHash,
|
||||
) GroupDatabase {
|
||||
return &groupDatabase{
|
||||
groupDB: groupDB,
|
||||
groupMemberDB: groupMemberDB,
|
||||
groupRequestDB: groupRequestDB,
|
||||
ctxTx: ctxTx,
|
||||
cache: redis2.NewGroupCacheRedis(rdb, localCache, groupDB, groupMemberDB, groupRequestDB, groupHash, redis2.GetRocksCacheOptions()),
|
||||
groupDB: groupDB,
|
||||
groupMemberDB: groupMemberDB,
|
||||
groupRequestDB: groupRequestDB,
|
||||
groupPinnedMsgDB: groupPinnedMsgDB,
|
||||
ctxTx: ctxTx,
|
||||
cache: redis2.NewGroupCacheRedis(rdb, localCache, groupDB, groupMemberDB, groupRequestDB, groupHash, redis2.GetRocksCacheOptions()),
|
||||
}
|
||||
}
|
||||
|
||||
type groupDatabase struct {
|
||||
groupDB database.Group
|
||||
groupMemberDB database.GroupMember
|
||||
groupRequestDB database.GroupRequest
|
||||
ctxTx tx.Tx
|
||||
cache cache.GroupCache
|
||||
groupDB database.Group
|
||||
groupMemberDB database.GroupMember
|
||||
groupRequestDB database.GroupRequest
|
||||
groupPinnedMsgDB database.GroupPinnedMsg
|
||||
ctxTx tx.Tx
|
||||
cache cache.GroupCache
|
||||
}
|
||||
|
||||
func (g *groupDatabase) PinGroupMessage(ctx context.Context, groupID string, msg *model.GroupPinnedMessage) ([]*model.GroupPinnedMessage, error) {
|
||||
return g.groupPinnedMsgDB.Pin(ctx, groupID, msg)
|
||||
}
|
||||
|
||||
func (g *groupDatabase) UnpinGroupMessage(ctx context.Context, groupID string, pinID string, seq int64) ([]*model.GroupPinnedMessage, error) {
|
||||
return g.groupPinnedMsgDB.Unpin(ctx, groupID, pinID, seq)
|
||||
}
|
||||
|
||||
func (g *groupDatabase) GetGroupPinnedMessages(ctx context.Context, groupID string) ([]*model.GroupPinnedMessage, error) {
|
||||
return g.groupPinnedMsgDB.Get(ctx, groupID)
|
||||
}
|
||||
|
||||
func (g *groupDatabase) FindJoinGroupID(ctx context.Context, userID string) ([]string, error) {
|
||||
|
||||
22
pkg/common/storage/database/group_pinned_msg.go
Normal file
22
pkg/common/storage/database/group_pinned_msg.go
Normal file
@ -0,0 +1,22 @@
|
||||
// Copyright © 2026 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.
|
||||
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
)
|
||||
|
||||
// GroupPinnedMsg 群置顶消息的存储抽象
|
||||
type GroupPinnedMsg interface {
|
||||
// Pin 置顶一条消息:若 PinID 为空会自动生成;自动滚动保留最近 N 条
|
||||
Pin(ctx context.Context, groupID string, msg *model.GroupPinnedMessage) ([]*model.GroupPinnedMessage, error)
|
||||
// Unpin 取消置顶;pinID 非空时按 pinID 精确删除,否则按 seq 删除
|
||||
Unpin(ctx context.Context, groupID string, pinID string, seq int64) ([]*model.GroupPinnedMessage, error)
|
||||
// Get 获取群置顶消息列表(最新的在前)
|
||||
Get(ctx context.Context, groupID string) ([]*model.GroupPinnedMessage, error)
|
||||
}
|
||||
115
pkg/common/storage/database/mgo/group_pinned_msg.go
Normal file
115
pkg/common/storage/database/mgo/group_pinned_msg.go
Normal file
@ -0,0 +1,115 @@
|
||||
// Copyright © 2026 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.
|
||||
|
||||
package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/mongoutil"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
func NewGroupPinnedMsgMongo(db *mongo.Database) (database.GroupPinnedMsg, error) {
|
||||
coll := db.Collection(database.GroupPinnedMsgName)
|
||||
_, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
|
||||
Keys: bson.D{{Key: "group_id", Value: 1}},
|
||||
Options: options.Index().SetUnique(true),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
return &groupPinnedMsgMgo{coll: coll}, nil
|
||||
}
|
||||
|
||||
type groupPinnedMsgMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (g *groupPinnedMsgMgo) get(ctx context.Context, groupID string) (*model.GroupPinnedMsg, error) {
|
||||
doc, err := mongoutil.FindOne[*model.GroupPinnedMsg](ctx, g.coll, bson.M{"group_id": groupID})
|
||||
if err != nil {
|
||||
if errs.ErrRecordNotFound.Is(err) || errors.Is(err, mongo.ErrNoDocuments) {
|
||||
return &model.GroupPinnedMsg{GroupID: groupID}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return doc, nil
|
||||
}
|
||||
|
||||
func (g *groupPinnedMsgMgo) Get(ctx context.Context, groupID string) ([]*model.GroupPinnedMessage, error) {
|
||||
doc, err := g.get(ctx, groupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return doc.PinnedMsgs, nil
|
||||
}
|
||||
|
||||
// Pin 置顶一条消息:
|
||||
// - 若提供的 msg.PinID 为空,则自动生成 ObjectID().Hex()
|
||||
// - 同 seq 的旧记录会被先移除避免重复
|
||||
// - 新记录 push 到数组首位,自动滚动保留最近 GroupPinnedMsgMaxKeep 条
|
||||
func (g *groupPinnedMsgMgo) Pin(ctx context.Context, groupID string, msg *model.GroupPinnedMessage) ([]*model.GroupPinnedMessage, error) {
|
||||
if msg == nil {
|
||||
return nil, errs.ErrArgs.WrapMsg("pin msg is nil")
|
||||
}
|
||||
if msg.PinID == "" {
|
||||
msg.PinID = primitive.NewObjectID().Hex()
|
||||
}
|
||||
msg.GroupID = groupID
|
||||
|
||||
if _, err := mongoutil.UpdateOneResult(ctx, g.coll,
|
||||
bson.M{"group_id": groupID},
|
||||
bson.M{"$pull": bson.M{"pinned_msgs": bson.M{"seq": msg.Seq}}},
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
filter := bson.M{"group_id": groupID}
|
||||
update := bson.M{
|
||||
"$push": bson.M{
|
||||
"pinned_msgs": bson.M{
|
||||
"$each": bson.A{msg},
|
||||
"$position": 0,
|
||||
"$slice": model.GroupPinnedMsgMaxKeep,
|
||||
},
|
||||
},
|
||||
"$setOnInsert": bson.M{"group_id": groupID},
|
||||
}
|
||||
opts := options.Update().SetUpsert(true)
|
||||
if _, err := g.coll.UpdateOne(ctx, filter, update, opts); err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
return g.Get(ctx, groupID)
|
||||
}
|
||||
|
||||
// Unpin 取消置顶:
|
||||
// - pinID 非空时按 pinID 精确删除(推荐)
|
||||
// - 否则按 seq 删除
|
||||
// 返回更新后的置顶列表(可能为空数组)
|
||||
func (g *groupPinnedMsgMgo) Unpin(ctx context.Context, groupID string, pinID string, seq int64) ([]*model.GroupPinnedMessage, error) {
|
||||
if pinID == "" && seq <= 0 {
|
||||
return nil, errs.ErrArgs.WrapMsg("either pinID or seq must be provided")
|
||||
}
|
||||
pull := bson.M{}
|
||||
if pinID != "" {
|
||||
pull["pin_id"] = pinID
|
||||
} else {
|
||||
pull["seq"] = seq
|
||||
}
|
||||
if _, err := mongoutil.UpdateOneResult(ctx, g.coll,
|
||||
bson.M{"group_id": groupID},
|
||||
bson.M{"$pull": bson.M{"pinned_msgs": pull}},
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return g.Get(ctx, groupID)
|
||||
}
|
||||
@ -12,6 +12,7 @@ const (
|
||||
GroupJoinVersionName = "group_join_version"
|
||||
ConversationVersionName = "conversation_version"
|
||||
GroupRequestName = "group_request"
|
||||
GroupPinnedMsgName = "group_pinned_msg"
|
||||
LogName = "log"
|
||||
ObjectName = "s3"
|
||||
UserName = "user"
|
||||
|
||||
69
pkg/common/storage/model/group_pinned_msg.go
Normal file
69
pkg/common/storage/model/group_pinned_msg.go
Normal file
@ -0,0 +1,69 @@
|
||||
// Copyright © 2026 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.
|
||||
|
||||
package model
|
||||
|
||||
// GroupPinnedMsgMaxKeep 群置顶消息最多保留的条数(最新置顶的在最前)
|
||||
const GroupPinnedMsgMaxKeep = 3
|
||||
|
||||
// GroupPinnedOfflinePush 离线推送信息快照
|
||||
type GroupPinnedOfflinePush struct {
|
||||
Title string `bson:"title"`
|
||||
Desc string `bson:"desc"`
|
||||
Ex string `bson:"ex"`
|
||||
IOSPushSound string `bson:"ios_push_sound"`
|
||||
IOSBadgeCount bool `bson:"ios_badge_count"`
|
||||
SignalInfo string `bson:"signal_info"`
|
||||
}
|
||||
|
||||
// GroupPinnedMessage 一条群置顶消息的完整内容快照
|
||||
// 置顶时把消息整体快照入库,避免后续消息删除/撤回影响已置顶展示
|
||||
type GroupPinnedMessage struct {
|
||||
// PinID 全局唯一 id,用于精准取消置顶(生产由 mongo ObjectID().Hex() 生成)
|
||||
PinID string `bson:"pin_id"`
|
||||
|
||||
// 会话 / 群信息
|
||||
ConversationID string `bson:"conversation_id"`
|
||||
GroupID string `bson:"group_id"`
|
||||
|
||||
// 消息标识
|
||||
Seq int64 `bson:"seq"`
|
||||
ServerMsgID string `bson:"server_msg_id"`
|
||||
ClientMsgID string `bson:"client_msg_id"`
|
||||
|
||||
// 发送方信息
|
||||
SendID string `bson:"send_id"`
|
||||
RecvID string `bson:"recv_id"`
|
||||
SenderPlatformID int32 `bson:"sender_platform_id"`
|
||||
SenderNickname string `bson:"sender_nickname"`
|
||||
SenderFaceURL string `bson:"sender_face_url"`
|
||||
|
||||
// 消息内容快照
|
||||
SessionType int32 `bson:"session_type"`
|
||||
MsgFrom int32 `bson:"msg_from"`
|
||||
ContentType int32 `bson:"content_type"`
|
||||
Content string `bson:"content"`
|
||||
AtUserIDList []string `bson:"at_user_id_list"`
|
||||
Options map[string]bool `bson:"options"`
|
||||
AttachedInfo string `bson:"attached_info"`
|
||||
Ex string `bson:"ex"`
|
||||
|
||||
OfflinePush *GroupPinnedOfflinePush `bson:"offline_push"`
|
||||
|
||||
// 时间
|
||||
SendTime int64 `bson:"send_time"`
|
||||
CreateTime int64 `bson:"create_time"`
|
||||
Status int32 `bson:"status"`
|
||||
|
||||
// 操作人 & 时间
|
||||
PinUserID string `bson:"pin_user_id"`
|
||||
PinTime int64 `bson:"pin_time"`
|
||||
}
|
||||
|
||||
// GroupPinnedMsg 一个群的置顶消息文档,按 group_id 唯一
|
||||
type GroupPinnedMsg struct {
|
||||
GroupID string `bson:"group_id"`
|
||||
PinnedMsgs []*GroupPinnedMessage `bson:"pinned_msgs"`
|
||||
}
|
||||
@ -75,6 +75,29 @@ func (x *MsgClient) GetActiveConversation(ctx context.Context, conversationIDs [
|
||||
return extractField(ctx, x.MsgClient.GetActiveConversation, req, (*msg.GetActiveConversationResp).GetConversations)
|
||||
}
|
||||
|
||||
// GetSingleMsgBySeq 根据会话 ID 与 seq 拉取一条消息(不存在时返回 nil)
|
||||
func (x *MsgClient) GetSingleMsgBySeq(ctx context.Context, conversationID string, seq int64) (*sdkws.MsgData, error) {
|
||||
if conversationID == "" || seq <= 0 {
|
||||
return nil, nil
|
||||
}
|
||||
req := &msg.GetMsgByConversationIDsReq{
|
||||
ConversationIDs: []string{conversationID},
|
||||
MaxSeqs: map[string]int64{conversationID: seq},
|
||||
}
|
||||
resp, err := x.MsgClient.GetMsgByConversationIDs(ctx, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m := resp.GetMsgDatas()
|
||||
if len(m) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
if v, ok := m[conversationID]; ok && v != nil && v.Seq == seq {
|
||||
return v, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (x *MsgClient) GetSeqMessage(ctx context.Context, userID string, conversations []*msg.ConversationSeqs) (map[string]*sdkws.PullMsgs, error) {
|
||||
if len(conversations) == 0 {
|
||||
return nil, nil
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user