feat: 对接 PostgreSQL 存储层

backend/internal/api/conversation.go

@@ -12,19 +12,23 @@ import (
 	apperr "github.com/hhs/camtalk/internal/errors"
 	"github.com/hhs/camtalk/internal/models"
 	"github.com/hhs/camtalk/internal/session"
+	"github.com/hhs/camtalk/internal/store"
 )
 
 // ConversationHandler 提供对话相关的 REST 端点。
 type ConversationHandler struct {
 	sessionMgr session.Manager
 	tokenMgr   *auth.TokenManager
+	msgRepo    store.MessageRepository // 可选，为 nil 时 fallback 到内存查询
 }
 
 // NewConversationHandler 创建 ConversationHandler。
-func NewConversationHandler(sessionMgr session.Manager, tokenMgr *auth.TokenManager) *ConversationHandler {
+// msgRepo 可选，为 nil 时消息查询走内存。
+func NewConversationHandler(sessionMgr session.Manager, tokenMgr *auth.TokenManager, msgRepo store.MessageRepository) *ConversationHandler {
 	return &ConversationHandler{
 		sessionMgr: sessionMgr,
 		tokenMgr:   tokenMgr,
+		msgRepo:    msgRepo,
 	}
 }
 
@@ -215,7 +219,7 @@ func (h *ConversationHandler) Delete(c *gin.Context) {
 //
 // 查询参数：
 //   - limit: 返回消息数量上限，默认 50
-//   - before: 消息偏移量（用于分页），返回此偏移量之前的消息
+//   - before: 消息 ID 游标（用于分页），返回此 ID 之前的消息
 func (h *ConversationHandler) GetMessages(c *gin.Context) {
 	sessionID := c.Param("id")
 
@@ -229,9 +233,27 @@ func (h *ConversationHandler) GetMessages(c *gin.Context) {
 		limit = 50
 	}
 
-	before, _ := strconv.Atoi(c.DefaultQuery("before", "0"))
+	beforeID, _ := strconv.ParseInt(c.DefaultQuery("before", "0"), 10, 64)
 
-	// 获取全量历史（内存实现中 history 是全量存储的）
+	// 优先从 PostgreSQL 查询（支持持久化后的全量历史）
+	if h.msgRepo != nil {
+		messages, err := h.msgRepo.GetMessages(c.Request.Context(), sessionID, limit, beforeID)
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{
+				"code":    apperr.CodeInternalError,
+				"message": "failed to get messages",
+			})
+			return
+		}
+		count, _ := h.msgRepo.GetMessageCount(c.Request.Context(), sessionID)
+		c.JSON(http.StatusOK, gin.H{
+			"messages": messages,
+			"total":    count,
+		})
+		return
+	}
+
+	// fallback：从内存查询
 	allMessages, err := h.sessionMgr.GetHistory(c.Request.Context(), sessionID, 0)
 	if err != nil {
 		if errors.Is(err, session.ErrSessionNotFound) {
@@ -250,9 +272,9 @@ func (h *ConversationHandler) GetMessages(c *gin.Context) {
 
 	total := len(allMessages)
 
-	// before > 0 表示取 before 之前的消息（不含 before 位置）
-	if before > 0 && before <= total {
-		allMessages = allMessages[:before]
+	// beforeID > 0 时表示偏移量（兼容旧接口语义）
+	if beforeID > 0 && int(beforeID) <= total {
+		allMessages = allMessages[:beforeID]
 	}
 
 	// 取最后 limit 条

backend/internal/api/conversation_test.go

@@ -93,7 +93,7 @@ func newConvTestRouter(mgr session.Manager) (*gin.Engine, *auth.TokenManager) {
 	gin.SetMode(gin.TestMode)
 	r := gin.New()
 	tm := auth.NewTokenManager("test-secret", 15*time.Minute, 7*24*time.Hour)
-	h := api.NewConversationHandler(mgr, tm)
+	h := api.NewConversationHandler(mgr, tm, nil)
 	h.RegisterRoutes(r.Group("/api"))
 	return r, tm
 }

backend/internal/store/migrate.go

@@ -0,0 +1,80 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"sort"
+	"strings"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/hhs/camtalk/internal/logger"
+)
+
+// RunMigrations 从给定的 fs.FS 中读取 *.up.sql 文件并按版本号顺序执行。
+// 已执行过的版本会跳过（通过 schema_migrations 表记录）。
+func RunMigrations(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS) error {
+	// 确保 schema_migrations 表存在
+	if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
+		version INTEGER PRIMARY KEY,
+		applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+	)`); err != nil {
+		return fmt.Errorf("create schema_migrations table: %w", err)
+	}
+
+	// 收集所有 *.up.sql 文件
+	entries, err := fs.ReadDir(fsys, ".")
+	if err != nil {
+		return fmt.Errorf("read migrations dir: %w", err)
+	}
+
+	var files []string
+	for _, e := range entries {
+		if !e.IsDir() && strings.HasSuffix(e.Name(), ".up.sql") {
+			files = append(files, e.Name())
+		}
+	}
+	sort.Strings(files)
+
+	for _, name := range files {
+		// 从文件名提取版本号，如 "001_users.up.sql" → 1
+		var version int
+		if _, err := fmt.Sscanf(name, "%d_", &version); err != nil {
+			return fmt.Errorf("parse version from %s: %w", name, err)
+		}
+
+		// 检查是否已执行
+		var exists bool
+		if err := pool.QueryRow(ctx,
+			`SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, version,
+		).Scan(&exists); err != nil {
+			return fmt.Errorf("check migration version %d: %w", version, err)
+		}
+		if exists {
+			logger.Log.Debugw("migration already applied", "version", version, "file", name)
+			continue
+		}
+
+		// 读取并执行
+		content, err := fs.ReadFile(fsys, name)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", name, err)
+		}
+
+		if _, err := pool.Exec(ctx, string(content)); err != nil {
+			return fmt.Errorf("execute migration %s: %w", name, err)
+		}
+
+		// 记录已执行
+		if _, err := pool.Exec(ctx,
+			`INSERT INTO schema_migrations (version) VALUES ($1)`, version,
+		); err != nil {
+			return fmt.Errorf("record migration %d: %w", version, err)
+		}
+
+		logger.Log.Infow("migration applied", "version", version, "file", name)
+	}
+
+	return nil
+}