CamTalk/backend/internal/store/session_pg.go

package store

import (
	"context"
	"errors"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/hhs/camtalk/internal/trace"
)

// 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 {
	log := trace.FromContext(ctx)

	_, 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,
	)
	if err != nil {
		log.Errorw("save session failed", "session_id", s.ID, "error", err)
		return err
	}

	log.Debugw("session saved", "session_id", s.ID, "user_id", s.UserID)
	return nil
}

func (r *PgSessionRepository) FindByID(ctx context.Context, id string) (*SessionRecord, error) {
	log := trace.FromContext(ctx)

	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 {
		log.Errorw("find session failed", "session_id", id, "error", err)
		return nil, err
	}

	log.Debugw("session found", "session_id", id)
	return &s, nil
}

func (r *PgSessionRepository) FindByUser(ctx context.Context, userID string, page, size int) ([]SessionRecord, int, error) {
	log := trace.FromContext(ctx)

	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 {
		log.Errorw("count user sessions failed", "user_id", userID, "error", err)
		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 {
		log.Errorw("find user sessions failed", "user_id", userID, "error", err)
		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 {
			log.Errorw("scan session row failed", "user_id", userID, "error", err)
			return nil, 0, err
		}
		list = append(list, s)
	}
	if err := rows.Err(); err != nil {
		log.Errorw("iterate session rows failed", "user_id", userID, "error", err)
		return nil, 0, err
	}

	log.Debugw("user sessions found", "user_id", userID, "count", len(list), "total", total)
	return list, total, nil
}

func (r *PgSessionRepository) UpdateTitle(ctx context.Context, id string, title string) error {
	log := trace.FromContext(ctx)

	tag, err := r.pool.Exec(ctx,
		`UPDATE sessions SET title = $2, updated_at = NOW() WHERE id = $1`,
		id, title,
	)
	if err != nil {
		log.Errorw("update session title failed", "session_id", id, "error", err)
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrSessionNotFound
	}

	log.Debugw("session title updated", "session_id", id)
	return nil
}

func (r *PgSessionRepository) UpdateConfig(ctx context.Context, id string, configJSON []byte) error {
	log := trace.FromContext(ctx)

	tag, err := r.pool.Exec(ctx,
		`UPDATE sessions SET config = $2, updated_at = NOW() WHERE id = $1`,
		id, configJSON,
	)
	if err != nil {
		log.Errorw("update session config failed", "session_id", id, "error", err)
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrSessionNotFound
	}

	log.Debugw("session config updated", "session_id", id)
	return nil
}

func (r *PgSessionRepository) Touch(ctx context.Context, id string) error {
	log := trace.FromContext(ctx)

	tag, err := r.pool.Exec(ctx,
		`UPDATE sessions SET updated_at = NOW() WHERE id = $1`, id,
	)
	if err != nil {
		log.Errorw("touch session failed", "session_id", id, "error", err)
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrSessionNotFound
	}

	log.Debugw("session touched", "session_id", id)
	return nil
}

func (r *PgSessionRepository) Delete(ctx context.Context, id string) error {
	log := trace.FromContext(ctx)

	tag, err := r.pool.Exec(ctx,
		`DELETE FROM sessions WHERE id = $1`, id,
	)
	if err != nil {
		log.Errorw("delete session failed", "session_id", id, "error", err)
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrSessionNotFound
	}

	log.Debugw("session deleted", "session_id", id)
	return nil
}