fix: 单聊系统消息带回当前用户视角的对方

SYNC_INBOX 补 conversation_type、peer_id、group_id 和 last_read_msg_id。历史消息在 sender 为 0 时用 peer_id 定位对方。已读清未读时保留会话 ZSET 分数，避免被毫秒时间戳冲到游标后面。

Co-authored-by: Cursor <cursoragent@cursor.com>

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)

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
+}

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

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()

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"`
+	ConversationID   int64  `json:"conversation_id"`
-	Score          int64  `json:"score"`
+	ConversationType int    `json:"conversation_type,omitempty"`
-	Unread         int64  `json:"unread"`
+	PeerID           int64  `json:"peer_id,omitempty"`
-	Preview        []byte `json:"preview"`
+	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"`
 }

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
+}

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,
 	})

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
+	// Score 0 keeps the existing ZSET score (Snowflake msg id). last_msg_at is unix
-	if conv, err := s.chats.Get(ctx, conversationID); err == nil && conv != nil {
+	// millis and would move the dialog backward, hiding it from incremental SYNC_INBOX.
-		score = conv.LastMsgAt
+	if err := s.inbox.ClearUnread(ctx, userID, conversationID, 0); err != nil {
-	}
-	if err := s.inbox.ClearUnread(ctx, userID, conversationID, score); err != nil {
 		return errs.ErrInner.WithError("inbox", err)
 	}
 	return nil

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"`

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)
+	}
+}

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 {