SYNC_INBOX 补 conversation_type、peer_id、group_id 和 last_read_msg_id。历史消息在 sender 为 0 时用 peer_id 定位对方。已读清未读时保留会话 ZSET 分数,避免被毫秒时间戳冲到游标后面。 Co-authored-by: Cursor <cursoragent@cursor.com>
143 lines
4.4 KiB
Go
143 lines
4.4 KiB
Go
package model
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"spark-server/internal/module/conversation/entity"
|
||
"spark-server/internal/pkg/errs"
|
||
)
|
||
|
||
// ConversationMember maps table conversation_members.
|
||
type ConversationMember struct {
|
||
ID int64 `gorm:"column:id;primaryKey;autoIncrement;comment:行 id,表内自增,仅内部用;业务唯一键是 (conversation_id, user_id)"`
|
||
ConversationID int64 `gorm:"column:conversation_id;comment:会话 id"`
|
||
UserID int64 `gorm:"column:user_id;comment:成员 user_id"`
|
||
LastReadMsgID int64 `gorm:"column:last_read_msg_id;comment:该成员已读到的最大消息 id,未读真相;Redis 丢了可从此重建"`
|
||
JoinedAt time.Time `gorm:"column:joined_at;comment:加入会话时间"`
|
||
Muted int `gorm:"column:muted;comment:0=未静音 1=免打扰"`
|
||
}
|
||
|
||
func (ConversationMember) TableName() string { return "t_conversation_members" }
|
||
|
||
func (r *ConversationMember) entity() *entity.Member {
|
||
if r == nil {
|
||
return nil
|
||
}
|
||
m := &entity.Member{
|
||
ConversationID: r.ConversationID, UserID: r.UserID,
|
||
LastReadMsgID: r.LastReadMsgID, JoinedAtMS: r.JoinedAt.UnixMilli(),
|
||
}
|
||
m.Muted = r.Muted != 0
|
||
return m
|
||
}
|
||
|
||
func (m *Repo) ListMemberIDs(ctx context.Context, conversationID int64) ([]int64, error) {
|
||
if conversationID <= 0 {
|
||
return nil, errs.ErrInvalidArgument
|
||
}
|
||
db, err := m.session(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var rows []ConversationMember
|
||
if err := db.Where("conversation_id = ?", conversationID).Find(&rows).Error; err != nil {
|
||
return nil, mapDBErr(err)
|
||
}
|
||
ids := make([]int64, 0, len(rows))
|
||
for i := range rows {
|
||
ids = append(ids, rows[i].UserID)
|
||
}
|
||
return ids, nil
|
||
}
|
||
|
||
func (m *Repo) UpdateLastRead(ctx context.Context, conversationID, userID, msgID int64) error {
|
||
if conversationID <= 0 || userID <= 0 || msgID <= 0 {
|
||
return errs.ErrInvalidArgument
|
||
}
|
||
db, err := m.session(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
res := db.Model(&ConversationMember{}).
|
||
Where("conversation_id = ? AND user_id = ? AND last_read_msg_id < ?", conversationID, userID, msgID).
|
||
Update("last_read_msg_id", msgID)
|
||
if res.Error != nil {
|
||
return mapDBErr(res.Error)
|
||
}
|
||
if res.RowsAffected > 0 {
|
||
return nil
|
||
}
|
||
var n int64
|
||
if err := db.Model(&ConversationMember{}).
|
||
Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Count(&n).Error; err != nil {
|
||
return mapDBErr(err)
|
||
}
|
||
if n == 0 {
|
||
return errs.ErrNotFound
|
||
}
|
||
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
|
||
}
|
||
db, err := m.session(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var row ConversationMember
|
||
if err := db.Where("conversation_id = ? AND user_id = ?", conversationID, userID).Take(&row).Error; err != nil {
|
||
return nil, mapDBErr(err)
|
||
}
|
||
return row.entity(), nil
|
||
}
|