diff --git a/internal/api/group.go b/internal/api/group.go index 9a2ffda06..eec11353a 100644 --- a/internal/api/group.go +++ b/internal/api/group.go @@ -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) +} diff --git a/internal/api/router.go b/internal/api/router.go index accaae666..026f327d7 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -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 { diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index f829f07cf..d9d8fd427 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -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) diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index 95335b2fc..3b1a24688 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -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() { diff --git a/internal/rpc/group/pinned_msg.go b/internal/rpc/group/pinned_msg.go new file mode 100644 index 000000000..d28977111 --- /dev/null +++ b/internal/rpc/group/pinned_msg.go @@ -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 +} diff --git a/pkg/common/storage/controller/group.go b/pkg/common/storage/controller/group.go index 539f7dccc..f8cbb0933 100644 --- a/pkg/common/storage/controller/group.go +++ b/pkg/common/storage/controller/group.go @@ -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) { diff --git a/pkg/common/storage/database/group_pinned_msg.go b/pkg/common/storage/database/group_pinned_msg.go new file mode 100644 index 000000000..82a268b6f --- /dev/null +++ b/pkg/common/storage/database/group_pinned_msg.go @@ -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) +} diff --git a/pkg/common/storage/database/mgo/group_pinned_msg.go b/pkg/common/storage/database/mgo/group_pinned_msg.go new file mode 100644 index 000000000..478bc7ebc --- /dev/null +++ b/pkg/common/storage/database/mgo/group_pinned_msg.go @@ -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) +} diff --git a/pkg/common/storage/database/name.go b/pkg/common/storage/database/name.go index 100e6d112..0fd9b3b2e 100644 --- a/pkg/common/storage/database/name.go +++ b/pkg/common/storage/database/name.go @@ -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" diff --git a/pkg/common/storage/model/group_pinned_msg.go b/pkg/common/storage/model/group_pinned_msg.go new file mode 100644 index 000000000..a80780875 --- /dev/null +++ b/pkg/common/storage/model/group_pinned_msg.go @@ -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"` +} diff --git a/pkg/rpcli/msg.go b/pkg/rpcli/msg.go index e4d1ece6e..d439d0c12 100644 --- a/pkg/rpcli/msg.go +++ b/pkg/rpcli/msg.go @@ -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