feat: 构建用户模块,实现用户对话历史持久化,完善接口文档 #96
@@ -187,12 +187,13 @@ func (m *MemoryManager) UpdateTitle(_ context.Context, sessionID string, title s
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ListByUser 获取用户的对话列表(分页,按 UpdatedAt 降序)。
|
// ListByUser 获取用户的对话列表(分页,按 UpdatedAt 降序)。
|
||||||
func (m *MemoryManager) ListByUser(_ context.Context, userID string, page, size int) ([]ConversationSummary, int, error) {
|
// 若配置了 MessageRepository,消息统计从 PostgreSQL 聚合查询(更准确)。
|
||||||
|
func (m *MemoryManager) ListByUser(ctx context.Context, userID string, page, size int) ([]ConversationSummary, int, error) {
|
||||||
m.mu.RLock()
|
m.mu.RLock()
|
||||||
defer m.mu.RUnlock()
|
|
||||||
|
|
||||||
// 收集该用户的所有 session
|
// 收集该用户的所有 session
|
||||||
var list []ConversationSummary
|
var list []ConversationSummary
|
||||||
|
var sessionIDs []string
|
||||||
for _, entry := range m.sessions {
|
for _, entry := range m.sessions {
|
||||||
if entry.session.UserID != userID {
|
if entry.session.UserID != userID {
|
||||||
continue
|
continue
|
||||||
@@ -201,15 +202,32 @@ func (m *MemoryManager) ListByUser(_ context.Context, userID string, page, size
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
summary := ConversationSummary{
|
summary := ConversationSummary{
|
||||||
ID: entry.session.ID,
|
ID: entry.session.ID,
|
||||||
Title: entry.session.Title,
|
Title: entry.session.Title,
|
||||||
MessageCount: len(entry.history),
|
UpdatedAt: entry.lastActive,
|
||||||
UpdatedAt: entry.lastActive,
|
|
||||||
}
|
}
|
||||||
|
// 先用内存值填充,后续可能被 PG 统计覆盖
|
||||||
|
summary.MessageCount = len(entry.history)
|
||||||
if len(entry.history) > 0 {
|
if len(entry.history) > 0 {
|
||||||
summary.LastMessage = entry.history[len(entry.history)-1].Content
|
summary.LastMessage = entry.history[len(entry.history)-1].Content
|
||||||
}
|
}
|
||||||
list = append(list, summary)
|
list = append(list, summary)
|
||||||
|
sessionIDs = append(sessionIDs, entry.session.ID)
|
||||||
|
}
|
||||||
|
m.mu.RUnlock()
|
||||||
|
|
||||||
|
// 若配置了 msgRepo,从 PostgreSQL 获取更准确的消息统计
|
||||||
|
if m.msgRepo != nil && len(sessionIDs) > 0 {
|
||||||
|
if stats, err := m.msgRepo.GetSessionMessageStats(ctx, sessionIDs); err == nil {
|
||||||
|
for i := range list {
|
||||||
|
if s, ok := stats[list[i].ID]; ok {
|
||||||
|
list[i].LastMessage = s.LastMessage
|
||||||
|
list[i].MessageCount = s.MessageCount
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
logger.Log.Warnw("get session message stats failed, falling back to in-memory", "error", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 按 UpdatedAt 降序排序
|
// 按 UpdatedAt 降序排序
|
||||||
|
|||||||
@@ -27,6 +27,16 @@ type MessageRepository interface {
|
|||||||
|
|
||||||
// GetMessageCount 获取会话的消息总数。
|
// GetMessageCount 获取会话的消息总数。
|
||||||
GetMessageCount(ctx context.Context, sessionID string) (int, error)
|
GetMessageCount(ctx context.Context, sessionID string) (int, error)
|
||||||
|
|
||||||
|
// GetSessionMessageStats 批量查询多个会话的消息统计(last_message + message_count)。
|
||||||
|
// 返回的 map key 为 sessionID,仅包含有消息的会话。
|
||||||
|
GetSessionMessageStats(ctx context.Context, sessionIDs []string) (map[string]SessionMessageStats, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SessionMessageStats 单个会话的消息统计(SQL 聚合查询结果)。
|
||||||
|
type SessionMessageStats struct {
|
||||||
|
LastMessage string
|
||||||
|
MessageCount int
|
||||||
}
|
}
|
||||||
|
|
||||||
// StoredMessage 持久化消息模型(store 层)。
|
// StoredMessage 持久化消息模型(store 层)。
|
||||||
|
|||||||
@@ -118,3 +118,46 @@ func (r *PgMessageRepository) GetMessageCount(ctx context.Context, sessionID str
|
|||||||
}
|
}
|
||||||
return count, nil
|
return count, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (r *PgMessageRepository) GetSessionMessageStats(ctx context.Context, sessionIDs []string) (map[string]SessionMessageStats, error) {
|
||||||
|
if len(sessionIDs) == 0 {
|
||||||
|
return map[string]SessionMessageStats{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
rows, err := r.pool.Query(ctx,
|
||||||
|
`WITH stats AS (
|
||||||
|
SELECT session_id, COUNT(*) AS cnt
|
||||||
|
FROM messages
|
||||||
|
WHERE session_id = ANY($1)
|
||||||
|
GROUP BY session_id
|
||||||
|
),
|
||||||
|
last_msg AS (
|
||||||
|
SELECT DISTINCT ON (session_id) session_id, content
|
||||||
|
FROM messages
|
||||||
|
WHERE session_id = ANY($1)
|
||||||
|
ORDER BY session_id, id DESC
|
||||||
|
)
|
||||||
|
SELECT s.session_id, s.cnt, COALESCE(lm.content, '')
|
||||||
|
FROM stats s
|
||||||
|
LEFT JOIN last_msg lm ON lm.session_id = s.session_id`,
|
||||||
|
sessionIDs,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
result := make(map[string]SessionMessageStats)
|
||||||
|
for rows.Next() {
|
||||||
|
var sid string
|
||||||
|
var stats SessionMessageStats
|
||||||
|
if err := rows.Scan(&sid, &stats.MessageCount, &stats.LastMessage); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
result[sid] = stats
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return result, nil
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user