diff --git a/backend/internal/session/memory.go b/backend/internal/session/memory.go index bc300a3..bfb232c 100644 --- a/backend/internal/session/memory.go +++ b/backend/internal/session/memory.go @@ -187,12 +187,13 @@ func (m *MemoryManager) UpdateTitle(_ context.Context, sessionID string, title s } // 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() - defer m.mu.RUnlock() // 收集该用户的所有 session var list []ConversationSummary + var sessionIDs []string for _, entry := range m.sessions { if entry.session.UserID != userID { continue @@ -201,15 +202,32 @@ func (m *MemoryManager) ListByUser(_ context.Context, userID string, page, size continue } summary := ConversationSummary{ - ID: entry.session.ID, - Title: entry.session.Title, - MessageCount: len(entry.history), - UpdatedAt: entry.lastActive, + ID: entry.session.ID, + Title: entry.session.Title, + UpdatedAt: entry.lastActive, } + // 先用内存值填充,后续可能被 PG 统计覆盖 + summary.MessageCount = len(entry.history) if len(entry.history) > 0 { summary.LastMessage = entry.history[len(entry.history)-1].Content } 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 降序排序 diff --git a/backend/internal/store/message.go b/backend/internal/store/message.go index 629af20..8220b8d 100644 --- a/backend/internal/store/message.go +++ b/backend/internal/store/message.go @@ -27,6 +27,16 @@ type MessageRepository interface { // GetMessageCount 获取会话的消息总数。 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 层)。 diff --git a/backend/internal/store/message_pg.go b/backend/internal/store/message_pg.go index 8c1512a..0bc0997 100644 --- a/backend/internal/store/message_pg.go +++ b/backend/internal/store/message_pg.go @@ -118,3 +118,46 @@ func (r *PgMessageRepository) GetMessageCount(ctx context.Context, sessionID str } 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 +}