From e27ae8723ae7e5df0c96c9c125a85e4756824a33 Mon Sep 17 00:00:00 2001 From: "DESKTOP-L8OUL9F\\gfl0034" <1003614916@qq.com> Date: Mon, 19 Oct 2026 15:04:03 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=8D=95=E8=81=8A=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=B8=A6=E5=9B=9E=E5=BD=93=E5=89=8D=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E8=A7=86=E8=A7=92=E7=9A=84=E5=AF=B9=E6=96=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SYNC_INBOX 补 conversation_type、peer_id、group_id 和 last_read_msg_id。历史消息在 sender 为 0 时用 peer_id 定位对方。已读清未读时保留会话 ZSET 分数，避免被毫秒时间戳冲到游标后面。 Co-authored-by: Cursor --- internal/bootstrap/app.go | 3 ++ .../conversation/entity/conversation.go | 10 ++++ .../conversation/model/conversation_member.go | 46 +++++++++++++++++++ internal/module/conversation/model/memory.go | 26 +++++++++++ internal/module/inbox/entity/preview.go | 15 ++++-- internal/module/message/logic/history.go | 25 +++++++++- internal/module/message/logic/message_test.go | 13 ++++++ internal/module/message/logic/read.go | 8 ++-- internal/module/message/logic/view.go | 1 + internal/module/sync/logic/inbox_test.go | 32 +++++++++++++ internal/module/sync/logic/service.go | 42 +++++++++++++++++ 11 files changed, 211 insertions(+), 10 deletions(-) diff --git a/internal/bootstrap/app.go b/internal/bootstrap/app.go index 40c8923..1005335 100644 --- a/internal/bootstrap/app.go +++ b/internal/bootstrap/app.go @@ -205,6 +205,9 @@ func finishApp(cfg *Config, log *zap.Logger, logClose func() error, idgen *id.Ge app.deliver, deliverQueue = wireDeliver(cfg, log, rdb, sessions, inbox, mailbox) syncSvc := synclogic.NewService(inbox, mailbox) chat := wireChat(idgen, db) + if faces, ok := chat.chats.(synclogic.DialogFaces); ok { + syncSvc.BindFaces(faces) + } friends := wireFriends(idgen, db, users, deliverQueue, chat) groups := wireGroups(idgen, db, users, friends, deliverQueue, objects, chat) moderation := wireModeration(db, users, idgen) diff --git a/internal/module/conversation/entity/conversation.go b/internal/module/conversation/entity/conversation.go index b828895..fd3b519 100644 --- a/internal/module/conversation/entity/conversation.go +++ b/internal/module/conversation/entity/conversation.go @@ -20,3 +20,13 @@ type Conversation struct { LastMsgID int64 LastMsgAt int64 } + +// DialogFace is the viewer-relative identity of one conversation. +// PeerID is the other member of a P2P chat. GroupID is groups.id, not the conversation id. +type DialogFace struct { + ConversationID int64 + Type Type + PeerID int64 + GroupID int64 + LastReadMsgID int64 +} diff --git a/internal/module/conversation/model/conversation_member.go b/internal/module/conversation/model/conversation_member.go index dd980ea..a697d8d 100644 --- a/internal/module/conversation/model/conversation_member.go +++ b/internal/module/conversation/model/conversation_member.go @@ -80,6 +80,52 @@ func (m *Repo) UpdateLastRead(ctx context.Context, conversationID, userID, msgID return nil } +func (m *Repo) DialogFaces(ctx context.Context, viewerID int64, conversationIDs []int64) (map[int64]entity.DialogFace, error) { + out := map[int64]entity.DialogFace{} + if len(conversationIDs) == 0 { + return out, nil + } + db, err := m.session(ctx) + if err != nil { + return nil, err + } + type faceRow struct { + ID int64 `gorm:"column:id"` + Type int `gorm:"column:type"` + UserSmall int64 `gorm:"column:user_small"` + UserBig int64 `gorm:"column:user_big"` + GroupID int64 `gorm:"column:group_id"` + LastReadMsgID int64 `gorm:"column:last_read_msg_id"` + } + var rows []faceRow + err = db.Table("t_conversations AS c"). + Select("c.id, c.type, COALESCE(p.user_small, 0) AS user_small, COALESCE(p.user_big, 0) AS user_big, COALESCE(g.id, 0) AS group_id, COALESCE(mem.last_read_msg_id, 0) AS last_read_msg_id"). + Joins("LEFT JOIN t_p2p_conversations AS p ON p.conversation_id = c.id"). + Joins("LEFT JOIN t_groups AS g ON g.conversation_id = c.id"). + Joins("LEFT JOIN t_conversation_members AS mem ON mem.conversation_id = c.id AND mem.user_id = ?", viewerID). + Where("c.id IN ?", conversationIDs). + Scan(&rows).Error + if err != nil { + return nil, mapDBErr(err) + } + for _, row := range rows { + face := entity.DialogFace{ + ConversationID: row.ID, Type: entity.Type(row.Type), + GroupID: row.GroupID, LastReadMsgID: row.LastReadMsgID, + } + if face.Type == entity.TypeP2P { + switch { + case row.UserSmall == viewerID: + face.PeerID = row.UserBig + case row.UserBig == viewerID: + face.PeerID = row.UserSmall + } + } + out[row.ID] = face + } + return out, nil +} + func (m *Repo) GetMember(ctx context.Context, conversationID, userID int64) (*entity.Member, error) { if conversationID <= 0 || userID <= 0 { return nil, errs.ErrInvalidArgument diff --git a/internal/module/conversation/model/memory.go b/internal/module/conversation/model/memory.go index 97e2778..b5e0a11 100644 --- a/internal/module/conversation/model/memory.go +++ b/internal/module/conversation/model/memory.go @@ -95,6 +95,32 @@ func (m *MemoryP2P) ListMemberIDs(_ context.Context, conversationID int64) ([]in return ids, nil } +func (m *MemoryP2P) DialogFaces(_ context.Context, viewerID int64, conversationIDs []int64) (map[int64]entity.DialogFace, error) { + m.mu.Lock() + defer m.mu.Unlock() + out := make(map[int64]entity.DialogFace, len(conversationIDs)) + for _, id := range conversationIDs { + conv, ok := m.byID[id] + if !ok { + continue + } + face := entity.DialogFace{ConversationID: id, Type: conv.Type} + if mem := m.members[id][viewerID]; mem != nil { + face.LastReadMsgID = mem.LastReadMsgID + } + if conv.Type == entity.TypeP2P { + for uid := range m.members[id] { + if uid > 0 && uid != viewerID { + face.PeerID = uid + break + } + } + } + out[id] = face + } + return out, nil +} + func (m *MemoryP2P) GetMember(_ context.Context, conversationID, userID int64) (*entity.Member, error) { m.mu.Lock() defer m.mu.Unlock() diff --git a/internal/module/inbox/entity/preview.go b/internal/module/inbox/entity/preview.go index 46978f5..53450e2 100644 --- a/internal/module/inbox/entity/preview.go +++ b/internal/module/inbox/entity/preview.go @@ -13,9 +13,16 @@ type DialogHit struct { } // Dialog is one conversation row for SYNC_INBOX. +// PeerID is the other user in a P2P chat (conversation_type=1), relative to the viewer. +// GroupID is groups.id (群号) when conversation_type=2. It is not conversation_id. +// LastReadMsgID is that viewer's member cursor; 0 means nothing read. type Dialog struct { - ConversationID int64 `json:"conversation_id"` - Score int64 `json:"score"` - Unread int64 `json:"unread"` - Preview []byte `json:"preview"` + ConversationID int64 `json:"conversation_id"` + ConversationType int `json:"conversation_type,omitempty"` + PeerID int64 `json:"peer_id,omitempty"` + GroupID int64 `json:"group_id,omitempty"` + LastReadMsgID int64 `json:"last_read_msg_id,omitempty"` + Score int64 `json:"score"` + Unread int64 `json:"unread"` + Preview []byte `json:"preview"` } diff --git a/internal/module/message/logic/history.go b/internal/module/message/logic/history.go index 6755e6c..324f92c 100644 --- a/internal/module/message/logic/history.go +++ b/internal/module/message/logic/history.go @@ -4,6 +4,7 @@ import ( "context" "errors" + conventity "spark-server/internal/module/conversation/entity" "spark-server/internal/pkg/errs" ) @@ -43,9 +44,31 @@ func (s *Service) GetHistory(ctx context.Context, userID int64, req HistoryReque if hasMore { rows = rows[:limit] } + peer := s.viewerPeer(ctx, userID, req.ConversationID) items := make([]*Item, 0, len(rows)) for _, row := range rows { - items = append(items, itemOf(row)) + item := itemOf(row) + item.PeerID = peer + items = append(items, item) } return &HistoryResponse{Items: items, HasMore: hasMore}, nil } + +// viewerPeer is the other member of a P2P conversation. Group chats stay 0; use group_id. +// System messages keep sender_id=0 and receiver_id=0; this field is how the viewer finds the peer. +func (s *Service) viewerPeer(ctx context.Context, viewerID, conversationID int64) int64 { + conv, err := s.chats.Get(ctx, conversationID) + if err != nil || conv == nil || conv.Type != conventity.TypeP2P { + return 0 + } + ids, err := s.chats.ListMemberIDs(ctx, conversationID) + if err != nil { + return 0 + } + for _, id := range ids { + if id > 0 && id != viewerID { + return id + } + } + return 0 +} diff --git a/internal/module/message/logic/message_test.go b/internal/module/message/logic/message_test.go index f1166f7..4651a5e 100644 --- a/internal/module/message/logic/message_test.go +++ b/internal/module/message/logic/message_test.go @@ -106,14 +106,27 @@ func TestSendReplyRecallDelete(t *testing.T) { t.Fatalf("history %+v %v", hist, err) } foundKind := false + foundSystem := false for _, it := range hist.Items { + if it.PeerID != b.ID { + t.Fatalf("viewer peer %d", it.PeerID) + } if it.MsgID == first.MsgID && it.Kind == msgentity.KindPrivate { foundKind = true } + if it.SenderID == 0 && it.MessageType == msgentity.TypeSystem { + foundSystem = true + if it.ReceiverID != 0 { + t.Fatalf("system receiver %d", it.ReceiverID) + } + } } if !foundKind { t.Fatal("private_message kind") } + if !foundSystem { + t.Fatal("system peer") + } caught, err := svc.GetHistory(ctx, a.ID, HistoryRequest{ ConversationID: msg.ConversationID, LastMsgID: hist.Items[len(hist.Items)-1].MsgID, }) diff --git a/internal/module/message/logic/read.go b/internal/module/message/logic/read.go index 84e546a..ac1d618 100644 --- a/internal/module/message/logic/read.go +++ b/internal/module/message/logic/read.go @@ -91,11 +91,9 @@ func (s *Service) clearDialogUnread(ctx context.Context, userID, conversationID if s.inbox == nil { return nil } - var score int64 - if conv, err := s.chats.Get(ctx, conversationID); err == nil && conv != nil { - score = conv.LastMsgAt - } - if err := s.inbox.ClearUnread(ctx, userID, conversationID, score); err != nil { + // Score 0 keeps the existing ZSET score (Snowflake msg id). last_msg_at is unix + // millis and would move the dialog backward, hiding it from incremental SYNC_INBOX. + if err := s.inbox.ClearUnread(ctx, userID, conversationID, 0); err != nil { return errs.ErrInner.WithError("inbox", err) } return nil diff --git a/internal/module/message/logic/view.go b/internal/module/message/logic/view.go index d9db870..90c5b8c 100644 --- a/internal/module/message/logic/view.go +++ b/internal/module/message/logic/view.go @@ -13,6 +13,7 @@ type Item struct { ConversationID int64 `json:"conversation_id"` SenderID int64 `json:"sender_id"` ReceiverID int64 `json:"receiver_id,omitempty"` + PeerID int64 `json:"peer_id,omitempty"` GroupID int64 `json:"group_id,omitempty"` ClientMsgID string `json:"client_msg_id,omitempty"` MessageType entity.Type `json:"message_type"` diff --git a/internal/module/sync/logic/inbox_test.go b/internal/module/sync/logic/inbox_test.go index 9d4eaea..3dcf311 100644 --- a/internal/module/sync/logic/inbox_test.go +++ b/internal/module/sync/logic/inbox_test.go @@ -4,6 +4,8 @@ import ( "testing" "spark-server/internal/infrastructure/id" + convlogic "spark-server/internal/module/conversation/logic" + convmodel "spark-server/internal/module/conversation/model" inboxmodel "spark-server/internal/module/inbox/model" "spark-server/internal/module/sync/entity" syncmodel "spark-server/internal/module/sync/model" @@ -43,3 +45,33 @@ func TestSyncInboxIncremental(t *testing.T) { t.Fatalf("mailbox %+v %v", rec, err) } } + +func TestSyncInboxPeerForSystemSender(t *testing.T) { + ids, err := id.New(1) + if err != nil { + t.Fatal(err) + } + ctx := t.Context() + chats := convmodel.NewMemoryP2P() + convID, err := convlogic.NewP2PService(chats, ids).EnsureP2P(ctx, 1001, 1002) + if err != nil { + t.Fatal(err) + } + if err := chats.UpdateLastRead(ctx, convID, 1001, 88001); err != nil { + t.Fatal(err) + } + inbox := inboxmodel.NewMemoryStore() + if err := inbox.TouchUnread(ctx, 1001, convID, 88001); err != nil { + t.Fatal(err) + } + svc := NewService(inbox, nil) + svc.BindFaces(chats) + out, err := svc.SyncInbox(ctx, entity.InboxRequest{UserID: 1001, DeviceID: 7}) + if err != nil || len(out.Dialogs) != 1 { + t.Fatalf("inbox %+v %v", out, err) + } + d := out.Dialogs[0] + if d.ConversationID != convID || d.ConversationType != 1 || d.PeerID != 1002 || d.GroupID != 0 || d.LastReadMsgID != 88001 { + t.Fatalf("face %+v", d) + } +} diff --git a/internal/module/sync/logic/service.go b/internal/module/sync/logic/service.go index ff780fa..0e0a340 100644 --- a/internal/module/sync/logic/service.go +++ b/internal/module/sync/logic/service.go @@ -4,6 +4,7 @@ package logic import ( "context" + conventity "spark-server/internal/module/conversation/entity" inboxentity "spark-server/internal/module/inbox/entity" inboxlogic "spark-server/internal/module/inbox/logic" "spark-server/internal/module/sync/entity" @@ -28,15 +29,28 @@ type InboxResponse struct { } // Service is SYNC_INBOX. HTTP and TCP share it. +// DialogFaces resolves viewer-relative peer / group / read cursor. Nil leaves those fields empty. +type DialogFaces interface { + DialogFaces(ctx context.Context, viewerID int64, conversationIDs []int64) (map[int64]conventity.DialogFace, error) +} + type Service struct { inbox inboxlogic.Store box Mailbox + faces DialogFaces } func NewService(inbox inboxlogic.Store, box Mailbox) *Service { return &Service{inbox: inbox, box: box} } +func (s *Service) BindFaces(faces DialogFaces) { + if s == nil { + return + } + s.faces = faces +} + func (s *Service) ready() bool { return s != nil && s.inbox != nil } @@ -67,6 +81,9 @@ func (s *Service) SyncInbox(ctx context.Context, req entity.InboxRequest) (*Inbo if dialogs == nil { dialogs = []inboxentity.Dialog{} } + if err := s.attachFaces(ctx, req.UserID, dialogs); err != nil { + return nil, err + } nextScore := maxScore(req.DialogScore, dialogs) mailbox, nextSeq, err := s.loadMailbox(ctx, req, limit) if err != nil { @@ -101,6 +118,31 @@ func (s *Service) loadMailbox(ctx context.Context, req entity.InboxRequest, limi return mailbox, nextSeq, nil } +func (s *Service) attachFaces(ctx context.Context, viewerID int64, dialogs []inboxentity.Dialog) error { + if s == nil || s.faces == nil || len(dialogs) == 0 { + return nil + } + ids := make([]int64, len(dialogs)) + for i := range dialogs { + ids[i] = dialogs[i].ConversationID + } + faces, err := s.faces.DialogFaces(ctx, viewerID, ids) + if err != nil { + return err + } + for i := range dialogs { + face, ok := faces[dialogs[i].ConversationID] + if !ok { + continue + } + dialogs[i].ConversationType = int(face.Type) + dialogs[i].PeerID = face.PeerID + dialogs[i].GroupID = face.GroupID + dialogs[i].LastReadMsgID = face.LastReadMsgID + } + return nil +} + func maxScore(cur int64, dialogs []inboxentity.Dialog) int64 { for i := range dialogs { if dialogs[i].Score > cur {