diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index 8accba9..cdca165 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -53,6 +53,7 @@ func main() { var userRepo store.UserRepository var msgRepo store.MessageRepository + var sessRepo store.SessionRepository if cfg.Storage.Driver == "postgres" { pool, err := store.NewPostgresPool(ctx, cfg.Storage.DSN) @@ -68,6 +69,7 @@ func main() { userRepo = store.NewPgUserRepository(pool) msgRepo = store.NewPgMessageRepository(pool) + sessRepo = store.NewPgSessionRepository(pool) logger.Log.Infow("postgres storage initialized", "driver", cfg.Storage.Driver) } else { userRepo = store.NewMemUserRepository() @@ -80,6 +82,9 @@ func main() { if msgRepo != nil { sessionOpts = append(sessionOpts, session.WithMessageRepository(msgRepo)) } + if sessRepo != nil { + sessionOpts = append(sessionOpts, session.WithSessionRepository(sessRepo)) + } sessionMgr = session.NewMemoryManager( time.Duration(cfg.Session.TTL)*time.Minute, cfg.Session.MaxHistory, diff --git a/backend/internal/session/memory.go b/backend/internal/session/memory.go index bfb232c..8ee3f3b 100644 --- a/backend/internal/session/memory.go +++ b/backend/internal/session/memory.go @@ -2,6 +2,7 @@ package session import ( "context" + "encoding/json" "sort" "sync" "time" @@ -34,7 +35,8 @@ type MemoryManager struct { ttl time.Duration maxHistory int stopCleaner chan struct{} - msgRepo store.MessageRepository // 可选，消息持久化（Write-Through） + msgRepo store.MessageRepository // 可选，消息持久化（Write-Through） + sessRepo store.SessionRepository // 可选，会话持久化（Write-Through） } // Option MemoryManager 的函数式选项。 @@ -47,6 +49,13 @@ func WithMessageRepository(repo store.MessageRepository) Option { } } +// WithSessionRepository 注入会话持久化仓库，启用会话元数据 Write-Through 模式。 +func WithSessionRepository(repo store.SessionRepository) Option { + return func(m *MemoryManager) { + m.sessRepo = repo + } +} + // NewMemoryManager 创建内存版 SessionManager。 // ttl 为会话过期时间，maxHistory 为对话历史上限（0 表示使用默认值 20）。 // opts 为可选配置，如 WithMessageRepository 启用消息持久化。 @@ -114,9 +123,8 @@ func (m *MemoryManager) isExpired(entry *sessionEntry) bool { } // Create 创建新会话。userID 为空表示匿名会话。 -func (m *MemoryManager) Create(_ context.Context, userID string, config models.SessionConfig) (string, error) { +func (m *MemoryManager) Create(ctx context.Context, userID string, config models.SessionConfig) (string, error) { m.mu.Lock() - defer m.mu.Unlock() id := uuid.New().String() now := time.Now() @@ -132,73 +140,181 @@ func (m *MemoryManager) Create(_ context.Context, userID string, config models.S history: make([]models.Message, 0), lastActive: now, } + m.mu.Unlock() + + // Write-Through：异步写 PG + if m.sessRepo != nil { + go func() { + cfgJSON, _ := json.Marshal(config) + if err := m.sessRepo.Save(ctx, store.SessionRecord{ + ID: id, UserID: userID, Title: models.DefaultSessionTitle, + Config: cfgJSON, CreatedAt: now, UpdatedAt: now, + }); err != nil { + logger.Log.Warnw("persist session failed", "session", id, "error", err) + } + }() + } logger.Log.Debugw("session created", "session", id, "user_id", userID) return id, nil } -// Get 获取会话。 -func (m *MemoryManager) Get(_ context.Context, sessionID string) (*models.Session, error) { +// Get 获取会话。内存中不存在时，尝试从 PG 加载（透明恢复）。 +func (m *MemoryManager) Get(ctx context.Context, sessionID string) (*models.Session, error) { m.mu.RLock() - defer m.mu.RUnlock() - entry, ok := m.sessions[sessionID] - if !ok || m.isExpired(entry) { - return nil, ErrSessionNotFound + if ok && !m.isExpired(entry) { + sess := entry.session + m.mu.RUnlock() + return &sess, nil + } + m.mu.RUnlock() + + // 内存未命中，尝试从 PG 加载 + if m.sessRepo != nil { + rec, err := m.sessRepo.FindByID(ctx, sessionID) + if err != nil { + return nil, ErrSessionNotFound + } + sess := m.recordToSession(rec) + // 加载到内存（含消息历史） + if m.msgRepo != nil { + _ = m.LoadSessionFromRepo(ctx, sess) + } else { + _ = m.LoadSession(sess, nil) + } + return sess, nil } - sess := entry.session // 复制一份返回 - return &sess, nil + return nil, ErrSessionNotFound } // UpdateConfig 更新会话配置。 -func (m *MemoryManager) UpdateConfig(_ context.Context, sessionID string, patch models.SessionConfigPatch) error { +func (m *MemoryManager) UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error { m.mu.Lock() - defer m.mu.Unlock() entry, ok := m.sessions[sessionID] if !ok || m.isExpired(entry) { + m.mu.Unlock() return ErrSessionNotFound } patch.Apply(&entry.session.Config) entry.lastActive = time.Now() + cfg := entry.session.Config + m.mu.Unlock() + + // Write-Through：异步更新 PG + if m.sessRepo != nil { + go func() { + cfgJSON, _ := json.Marshal(cfg) + if err := m.sessRepo.UpdateConfig(ctx, sessionID, cfgJSON); err != nil { + logger.Log.Warnw("update session config in DB failed", "session", sessionID, "error", err) + } + }() + } logger.Log.Debugw("session config updated", "session", sessionID) return nil } // UpdateTitle 更新会话标题。 -func (m *MemoryManager) UpdateTitle(_ context.Context, sessionID string, title string) error { +func (m *MemoryManager) UpdateTitle(ctx context.Context, sessionID string, title string) error { m.mu.Lock() - defer m.mu.Unlock() entry, ok := m.sessions[sessionID] if !ok || m.isExpired(entry) { + m.mu.Unlock() return ErrSessionNotFound } entry.session.Title = title entry.session.UpdatedAt = time.Now() entry.lastActive = time.Now() + m.mu.Unlock() + + // Write-Through：异步更新 PG + if m.sessRepo != nil { + go func() { + if err := m.sessRepo.UpdateTitle(ctx, sessionID, title); err != nil { + logger.Log.Warnw("update session title in DB failed", "session", sessionID, "error", err) + } + }() + } logger.Log.Debugw("session title updated", "session", sessionID, "title", title) return nil } // ListByUser 获取用户的对话列表（分页，按 UpdatedAt 降序）。 +// 若配置了 SessionRepository，从 PG 查询（包含内存中已过期的会话）。 // 若配置了 MessageRepository，消息统计从 PostgreSQL 聚合查询（更准确）。 func (m *MemoryManager) ListByUser(ctx context.Context, userID string, page, size int) ([]ConversationSummary, int, error) { + if page <= 0 { + page = 1 + } + if size <= 0 { + size = 20 + } + + // 优先从 PG 查询会话列表（包含已过期的会话） + if m.sessRepo != nil { + recs, total, err := m.sessRepo.FindByUser(ctx, userID, page, size) + if err != nil { + logger.Log.Warnw("list sessions from DB failed, falling back to in-memory", "error", err) + return m.listByUserFromMemory(ctx, userID, page, size) + } + + list := make([]ConversationSummary, 0, len(recs)) + var sessionIDs []string + for _, rec := range recs { + list = append(list, ConversationSummary{ + ID: rec.ID, + Title: rec.Title, + UpdatedAt: rec.UpdatedAt, + }) + sessionIDs = append(sessionIDs, rec.ID) + } + + // 用内存中的消息数填充 + m.mu.RLock() + for i := range list { + if entry, ok := m.sessions[list[i].ID]; ok { + list[i].MessageCount = len(entry.history) + if len(entry.history) > 0 { + list[i].LastMessage = entry.history[len(entry.history)-1].Content + } + } + } + m.mu.RUnlock() + + // 从 PG 获取更准确的消息统计 + 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 + } + } + } + } + + return list, total, nil + } + + // fallback：纯内存查询 + return m.listByUserFromMemory(ctx, userID, page, size) +} + +// listByUserFromMemory 从内存中获取用户的对话列表（无 PG 时的 fallback）。 +func (m *MemoryManager) listByUserFromMemory(ctx context.Context, userID string, page, size int) ([]ConversationSummary, int, error) { m.mu.RLock() - // 收集该用户的所有 session var list []ConversationSummary var sessionIDs []string for _, entry := range m.sessions { - if entry.session.UserID != userID { - continue - } - if m.isExpired(entry) { + if entry.session.UserID != userID || m.isExpired(entry) { continue } summary := ConversationSummary{ @@ -206,7 +322,6 @@ func (m *MemoryManager) ListByUser(ctx context.Context, userID string, page, siz 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 @@ -216,7 +331,7 @@ func (m *MemoryManager) ListByUser(ctx context.Context, userID string, page, siz } m.mu.RUnlock() - // 若配置了 msgRepo，从 PostgreSQL 获取更准确的消息统计 + // 从 PG 获取更准确的消息统计 if m.msgRepo != nil && len(sessionIDs) > 0 { if stats, err := m.msgRepo.GetSessionMessageStats(ctx, sessionIDs); err == nil { for i := range list { @@ -225,25 +340,14 @@ func (m *MemoryManager) ListByUser(ctx context.Context, userID string, page, siz list[i].MessageCount = s.MessageCount } } - } else { - logger.Log.Warnw("get session message stats failed, falling back to in-memory", "error", err) } } - // 按 UpdatedAt 降序排序 sort.Slice(list, func(i, j int) bool { return list[i].UpdatedAt.After(list[j].UpdatedAt) }) total := len(list) - - // 分页 - if page <= 0 { - page = 1 - } - if size <= 0 { - size = 20 - } start := (page - 1) * size if start >= total { return []ConversationSummary{}, total, nil @@ -425,15 +529,26 @@ func (m *MemoryManager) Touch(_ context.Context, sessionID string) error { } // Destroy 显式销毁会话。 -func (m *MemoryManager) Destroy(_ context.Context, sessionID string) error { +func (m *MemoryManager) Destroy(ctx context.Context, sessionID string) error { m.mu.Lock() - defer m.mu.Unlock() if _, ok := m.sessions[sessionID]; !ok { + m.mu.Unlock() return ErrSessionNotFound } delete(m.sessions, sessionID) + m.mu.Unlock() + + // Write-Through：异步删除 PG + if m.sessRepo != nil { + go func() { + if err := m.sessRepo.Delete(ctx, sessionID); err != nil { + logger.Log.Warnw("delete session from DB failed", "session", sessionID, "error", err) + } + }() + } + logger.Log.Debugw("session destroyed", "session", sessionID) return nil } @@ -452,3 +567,19 @@ func (m *MemoryManager) ActiveCount() int { } return count } + +// recordToSession 将 store.SessionRecord 转换为 models.Session。 +func (m *MemoryManager) recordToSession(rec *store.SessionRecord) *models.Session { + cfg := models.DefaultConfig() + if len(rec.Config) > 0 { + _ = json.Unmarshal(rec.Config, &cfg) + } + return &models.Session{ + ID: rec.ID, + UserID: rec.UserID, + Title: rec.Title, + CreatedAt: rec.CreatedAt, + UpdatedAt: rec.UpdatedAt, + Config: cfg, + } +} diff --git a/backend/internal/store/session.go b/backend/internal/store/session.go new file mode 100644 index 0000000..cbe8c9e --- /dev/null +++ b/backend/internal/store/session.go @@ -0,0 +1,47 @@ +package store + +import ( + "context" + "errors" + "time" +) + +var ( + // ErrSessionNotFound 会话不存在。 + ErrSessionNotFound = errors.New("session not found") +) + +// SessionRepository 会话持久化接口。 +type SessionRepository interface { + // Save 创建或更新会话（UPSERT）。 + Save(ctx context.Context, s SessionRecord) error + + // FindByID 根据 ID 查询会话。 + FindByID(ctx context.Context, id string) (*SessionRecord, error) + + // FindByUser 查询用户的会话列表（分页，按 updated_at 降序）。 + // 返回 (列表, 总数, error)。 + FindByUser(ctx context.Context, userID string, page, size int) ([]SessionRecord, int, error) + + // UpdateTitle 更新会话标题。 + UpdateTitle(ctx context.Context, id string, title string) error + + // UpdateConfig 更新会话配置。 + UpdateConfig(ctx context.Context, id string, configJSON []byte) error + + // Touch 刷新 updated_at。 + Touch(ctx context.Context, id string) error + + // Delete 删除会话。 + Delete(ctx context.Context, id string) error +} + +// SessionRecord 持久化会话模型（store 层）。 +type SessionRecord struct { + ID string + UserID string + Title string + Config []byte // JSON 编码的 SessionConfig + CreatedAt time.Time + UpdatedAt time.Time +} diff --git a/backend/internal/store/session_pg.go b/backend/internal/store/session_pg.go new file mode 100644 index 0000000..53aff40 --- /dev/null +++ b/backend/internal/store/session_pg.go @@ -0,0 +1,146 @@ +package store + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// PgSessionRepository 基于 PostgreSQL 的 SessionRepository 实现。 +type PgSessionRepository struct { + pool *pgxpool.Pool +} + +// NewPgSessionRepository 创建 PgSessionRepository。 +func NewPgSessionRepository(pool *pgxpool.Pool) *PgSessionRepository { + return &PgSessionRepository{pool: pool} +} + +func (r *PgSessionRepository) Save(ctx context.Context, s SessionRecord) error { + _, err := r.pool.Exec(ctx, + `INSERT INTO sessions (id, user_id, title, config, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (id) DO UPDATE SET + title = EXCLUDED.title, + config = EXCLUDED.config, + updated_at = EXCLUDED.updated_at`, + s.ID, s.UserID, s.Title, s.Config, s.CreatedAt, s.UpdatedAt, + ) + return err +} + +func (r *PgSessionRepository) FindByID(ctx context.Context, id string) (*SessionRecord, error) { + var s SessionRecord + err := r.pool.QueryRow(ctx, + `SELECT id, user_id, title, config, created_at, updated_at + FROM sessions WHERE id = $1`, id, + ).Scan(&s.ID, &s.UserID, &s.Title, &s.Config, &s.CreatedAt, &s.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrSessionNotFound + } + if err != nil { + return nil, err + } + return &s, nil +} + +func (r *PgSessionRepository) FindByUser(ctx context.Context, userID string, page, size int) ([]SessionRecord, int, error) { + if page <= 0 { + page = 1 + } + if size <= 0 { + size = 20 + } + offset := (page - 1) * size + + // 查询总数 + var total int + if err := r.pool.QueryRow(ctx, + `SELECT COUNT(*) FROM sessions WHERE user_id = $1`, userID, + ).Scan(&total); err != nil { + return nil, 0, err + } + + // 查询列表 + rows, err := r.pool.Query(ctx, + `SELECT id, user_id, title, config, created_at, updated_at + FROM sessions + WHERE user_id = $1 + ORDER BY updated_at DESC + LIMIT $2 OFFSET $3`, + userID, size, offset, + ) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var list []SessionRecord + for rows.Next() { + var s SessionRecord + if err := rows.Scan(&s.ID, &s.UserID, &s.Title, &s.Config, &s.CreatedAt, &s.UpdatedAt); err != nil { + return nil, 0, err + } + list = append(list, s) + } + if err := rows.Err(); err != nil { + return nil, 0, err + } + return list, total, nil +} + +func (r *PgSessionRepository) UpdateTitle(ctx context.Context, id string, title string) error { + tag, err := r.pool.Exec(ctx, + `UPDATE sessions SET title = $2, updated_at = NOW() WHERE id = $1`, + id, title, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrSessionNotFound + } + return nil +} + +func (r *PgSessionRepository) UpdateConfig(ctx context.Context, id string, configJSON []byte) error { + tag, err := r.pool.Exec(ctx, + `UPDATE sessions SET config = $2, updated_at = NOW() WHERE id = $1`, + id, configJSON, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrSessionNotFound + } + return nil +} + +func (r *PgSessionRepository) Touch(ctx context.Context, id string) error { + tag, err := r.pool.Exec(ctx, + `UPDATE sessions SET updated_at = NOW() WHERE id = $1`, id, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrSessionNotFound + } + return nil +} + +func (r *PgSessionRepository) Delete(ctx context.Context, id string) error { + tag, err := r.pool.Exec(ctx, + `DELETE FROM sessions WHERE id = $1`, id, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrSessionNotFound + } + return nil +} diff --git a/backend/migrations/003_sessions.down.sql b/backend/migrations/003_sessions.down.sql new file mode 100644 index 0000000..63d205d --- /dev/null +++ b/backend/migrations/003_sessions.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS sessions; diff --git a/backend/migrations/003_sessions.up.sql b/backend/migrations/003_sessions.up.sql new file mode 100644 index 0000000..0c3d7fa --- /dev/null +++ b/backend/migrations/003_sessions.up.sql @@ -0,0 +1,11 @@ +CREATE TABLE IF NOT EXISTS sessions ( + id UUID PRIMARY KEY, + user_id UUID NOT NULL, + title VARCHAR(256) NOT NULL DEFAULT '新对话', + config JSONB NOT NULL DEFAULT '{}', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions (user_id); +CREATE INDEX IF NOT EXISTS idx_sessions_user_updated ON sessions (user_id, updated_at DESC);