diff --git a/backend/internal/store/message_pg.go b/backend/internal/store/message_pg.go index 0bc0997..014194b 100644 --- a/backend/internal/store/message_pg.go +++ b/backend/internal/store/message_pg.go @@ -8,6 +8,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/trace" ) // PgMessageRepository 基于 PostgreSQL 的 MessageRepository 实现。 @@ -21,14 +22,24 @@ func NewPgMessageRepository(pool *pgxpool.Pool) *PgMessageRepository { } func (r *PgMessageRepository) SaveMessage(ctx context.Context, sessionID string, msg models.Message, tokensUsed int) error { + log := trace.FromContext(ctx) + _, err := r.pool.Exec(ctx, `INSERT INTO messages (session_id, role, content, tokens_used) VALUES ($1, $2, $3, $4)`, sessionID, msg.Role, msg.Content, tokensUsed, ) - return err + if err != nil { + log.Errorw("save message failed", "session_id", sessionID, "role", msg.Role, "error", err) + return err + } + + log.Debugw("message saved", "session_id", sessionID, "role", msg.Role, "tokens_used", tokensUsed) + return nil } func (r *PgMessageRepository) GetMessages(ctx context.Context, sessionID string, limit int, beforeID int64) ([]StoredMessage, error) { + log := trace.FromContext(ctx) + if limit <= 0 { limit = 50 } @@ -56,6 +67,7 @@ func (r *PgMessageRepository) GetMessages(ctx context.Context, sessionID string, ) } if err != nil { + log.Errorw("get messages failed", "session_id", sessionID, "error", err) return nil, err } @@ -64,12 +76,16 @@ func (r *PgMessageRepository) GetMessages(ctx context.Context, sessionID string, rows[i], rows[j] = rows[j], rows[i] } + log.Debugw("messages retrieved", "session_id", sessionID, "count", len(rows)) return rows, nil } func (r *PgMessageRepository) queryMessages(ctx context.Context, query string, args ...any) ([]StoredMessage, error) { + log := trace.FromContext(ctx) + pgxRows, err := r.pool.Query(ctx, query, args...) if err != nil { + log.Errorw("query messages failed", "error", err) return nil, err } defer pgxRows.Close() @@ -78,17 +94,21 @@ func (r *PgMessageRepository) queryMessages(ctx context.Context, query string, a for pgxRows.Next() { var m StoredMessage if err := pgxRows.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &m.TokensUsed, &m.CreatedAt); err != nil { + log.Errorw("scan message row failed", "error", err) return nil, err } messages = append(messages, m) } if err := pgxRows.Err(); err != nil { + log.Errorw("iterate message rows failed", "error", err) return nil, err } return messages, nil } func (r *PgMessageRepository) GetLastMessage(ctx context.Context, sessionID string) (*StoredMessage, error) { + log := trace.FromContext(ctx) + var m StoredMessage err := r.pool.QueryRow(ctx, `SELECT id, session_id, role, content, tokens_used, created_at @@ -102,24 +122,34 @@ func (r *PgMessageRepository) GetLastMessage(ctx context.Context, sessionID stri return nil, ErrMessageNotFound } if err != nil { + log.Errorw("get last message failed", "session_id", sessionID, "error", err) return nil, err } + + log.Debugw("last message retrieved", "session_id", sessionID, "message_id", m.ID) return &m, nil } func (r *PgMessageRepository) GetMessageCount(ctx context.Context, sessionID string) (int, error) { + log := trace.FromContext(ctx) + var count int err := r.pool.QueryRow(ctx, `SELECT COUNT(*) FROM messages WHERE session_id = $1`, sessionID, ).Scan(&count) if err != nil { + log.Errorw("get message count failed", "session_id", sessionID, "error", err) return 0, err } + + log.Debugw("message count retrieved", "session_id", sessionID, "count", count) return count, nil } func (r *PgMessageRepository) GetSessionMessageStats(ctx context.Context, sessionIDs []string) (map[string]SessionMessageStats, error) { + log := trace.FromContext(ctx) + if len(sessionIDs) == 0 { return map[string]SessionMessageStats{}, nil } @@ -143,6 +173,7 @@ func (r *PgMessageRepository) GetSessionMessageStats(ctx context.Context, sessio sessionIDs, ) if err != nil { + log.Errorw("get session message stats failed", "session_count", len(sessionIDs), "error", err) return nil, err } defer rows.Close() @@ -152,12 +183,16 @@ func (r *PgMessageRepository) GetSessionMessageStats(ctx context.Context, sessio var sid string var stats SessionMessageStats if err := rows.Scan(&sid, &stats.MessageCount, &stats.LastMessage); err != nil { + log.Errorw("scan message stats row failed", "error", err) return nil, err } result[sid] = stats } if err := rows.Err(); err != nil { + log.Errorw("iterate message stats rows failed", "error", err) return nil, err } + + log.Debugw("session message stats retrieved", "session_count", len(sessionIDs), "result_count", len(result)) return result, nil } diff --git a/backend/internal/store/session_pg.go b/backend/internal/store/session_pg.go index 53aff40..e0ce9a0 100644 --- a/backend/internal/store/session_pg.go +++ b/backend/internal/store/session_pg.go @@ -6,6 +6,8 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + + "github.com/hhs/camtalk/internal/trace" ) // PgSessionRepository 基于 PostgreSQL 的 SessionRepository 实现。 @@ -19,6 +21,8 @@ func NewPgSessionRepository(pool *pgxpool.Pool) *PgSessionRepository { } 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) @@ -28,10 +32,18 @@ func (r *PgSessionRepository) Save(ctx context.Context, s SessionRecord) error { updated_at = EXCLUDED.updated_at`, s.ID, s.UserID, s.Title, s.Config, s.CreatedAt, s.UpdatedAt, ) - return err + 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 @@ -41,12 +53,17 @@ func (r *PgSessionRepository) FindByID(ctx context.Context, id string) (*Session 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 } @@ -60,6 +77,7 @@ func (r *PgSessionRepository) FindByUser(ctx context.Context, userID string, pag 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 } @@ -73,6 +91,7 @@ func (r *PgSessionRepository) FindByUser(ctx context.Context, userID string, pag userID, size, offset, ) if err != nil { + log.Errorw("find user sessions failed", "user_id", userID, "error", err) return nil, 0, err } defer rows.Close() @@ -81,66 +100,90 @@ func (r *PgSessionRepository) FindByUser(ctx context.Context, userID string, pag 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 } diff --git a/backend/internal/store/user_pg.go b/backend/internal/store/user_pg.go index fec0631..89390ed 100644 --- a/backend/internal/store/user_pg.go +++ b/backend/internal/store/user_pg.go @@ -7,6 +7,8 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + + "github.com/hhs/camtalk/internal/trace" ) // PgUserRepository 基于 PostgreSQL 的 UserRepository 实现。 @@ -20,18 +22,25 @@ func NewPgUserRepository(pool *pgxpool.Pool) *PgUserRepository { } func (r *PgUserRepository) Create(ctx context.Context, username, passwordHash string) (string, error) { + log := trace.FromContext(ctx) + var id string err := r.pool.QueryRow(ctx, `INSERT INTO users (username, password_hash) VALUES ($1, $2) RETURNING id`, username, passwordHash, ).Scan(&id) if err != nil { + log.Errorw("create user failed", "username", username, "error", err) return "", err } + + log.Debugw("user created", "user_id", id, "username", username) return id, nil } func (r *PgUserRepository) FindByUsername(ctx context.Context, username string) (*User, error) { + log := trace.FromContext(ctx) + var u User err := r.pool.QueryRow(ctx, `SELECT id, username, password_hash, created_at, updated_at FROM users WHERE username = $1`, @@ -41,12 +50,17 @@ func (r *PgUserRepository) FindByUsername(ctx context.Context, username string) return nil, ErrUserNotFound } if err != nil { + log.Errorw("find user by username failed", "username", username, "error", err) return nil, err } + + log.Debugw("user found by username", "user_id", u.ID, "username", username) return &u, nil } func (r *PgUserRepository) FindByID(ctx context.Context, id string) (*User, error) { + log := trace.FromContext(ctx) + var u User err := r.pool.QueryRow(ctx, `SELECT id, username, password_hash, created_at, updated_at FROM users WHERE id = $1`, @@ -56,20 +70,33 @@ func (r *PgUserRepository) FindByID(ctx context.Context, id string) (*User, erro return nil, ErrUserNotFound } if err != nil { + log.Errorw("find user by id failed", "user_id", id, "error", err) return nil, err } + + log.Debugw("user found by id", "user_id", id) return &u, nil } func (r *PgUserRepository) SaveRefreshToken(ctx context.Context, userID, tokenHash string, expiresAt time.Time) error { + log := trace.FromContext(ctx) + _, err := r.pool.Exec(ctx, `INSERT INTO refresh_tokens (user_id, token_hash, expires_at) VALUES ($1, $2, $3)`, userID, tokenHash, expiresAt, ) - return err + if err != nil { + log.Errorw("save refresh token failed", "user_id", userID, "error", err) + return err + } + + log.Debugw("refresh token saved", "user_id", userID) + return nil } func (r *PgUserRepository) FindRefreshToken(ctx context.Context, tokenHash string) (string, error) { + log := trace.FromContext(ctx) + var userID string err := r.pool.QueryRow(ctx, `SELECT user_id FROM refresh_tokens WHERE token_hash = $1 AND expires_at > NOW()`, @@ -79,23 +106,42 @@ func (r *PgUserRepository) FindRefreshToken(ctx context.Context, tokenHash strin return "", ErrRefreshTokenNotFound } if err != nil { + log.Errorw("find refresh token failed", "error", err) return "", err } + + log.Debugw("refresh token found", "user_id", userID) return userID, nil } func (r *PgUserRepository) DeleteRefreshToken(ctx context.Context, tokenHash string) error { + log := trace.FromContext(ctx) + _, err := r.pool.Exec(ctx, `DELETE FROM refresh_tokens WHERE token_hash = $1`, tokenHash, ) - return err + if err != nil { + log.Errorw("delete refresh token failed", "error", err) + return err + } + + log.Debugw("refresh token deleted") + return nil } func (r *PgUserRepository) DeleteUserRefreshTokens(ctx context.Context, userID string) error { + log := trace.FromContext(ctx) + _, err := r.pool.Exec(ctx, `DELETE FROM refresh_tokens WHERE user_id = $1`, userID, ) - return err + if err != nil { + log.Errorw("delete user refresh tokens failed", "user_id", userID, "error", err) + return err + } + + log.Debugw("user refresh tokens deleted", "user_id", userID) + return nil } diff --git a/backend/internal/store/user_scenario_repository.go b/backend/internal/store/user_scenario_repository.go index ed97dc8..a1e0eb3 100644 --- a/backend/internal/store/user_scenario_repository.go +++ b/backend/internal/store/user_scenario_repository.go @@ -10,6 +10,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/trace" ) // UserScenarioRepository 用户自建情景仓储接口。 @@ -35,6 +36,8 @@ func NewPostgresUserScenarioRepo(pool *pgxpool.Pool) UserScenarioRepository { // Create 创建用户情景。 func (r *PostgresUserScenarioRepo) Create(ctx context.Context, scenario *models.UserScenario) error { + log := trace.FromContext(ctx) + query := ` INSERT INTO user_scenarios (id, user_id, name, icon, description, prompt, greeting, language, created_at, updated_at) VALUES ($1, $2, $3, $4, NULLIF($5, ''), $6, NULLIF($7, ''), $8, $9, $10) @@ -69,13 +72,18 @@ func (r *PostgresUserScenarioRepo) Create(ctx context.Context, scenario *models. ).Scan(&scenario.ID, &scenario.CreatedAt, &scenario.UpdatedAt) if err != nil { + log.Errorw("create user scenario failed", "user_id", scenario.UserID, "name", scenario.Name, "error", err) return fmt.Errorf("create user scenario: %w", err) } + + log.Debugw("user scenario created", "scenario_id", scenario.ID, "user_id", scenario.UserID, "name", scenario.Name) return nil } // FindByID 根据 ID 查找情景。 func (r *PostgresUserScenarioRepo) FindByID(ctx context.Context, id string) (*models.UserScenario, error) { + log := trace.FromContext(ctx) + query := ` SELECT id, user_id, name, icon, description, prompt, greeting, language, created_at, updated_at FROM user_scenarios @@ -100,13 +108,18 @@ func (r *PostgresUserScenarioRepo) FindByID(ctx context.Context, id string) (*mo return nil, fmt.Errorf("user scenario not found: %s", id) } if err != nil { + log.Errorw("find user scenario failed", "scenario_id", id, "error", err) return nil, fmt.Errorf("find user scenario: %w", err) } + + log.Debugw("user scenario found", "scenario_id", id) return &scenario, nil } // FindByIDAndUserID 根据 ID 和用户 ID 查找情景（权限校验）。 func (r *PostgresUserScenarioRepo) FindByIDAndUserID(ctx context.Context, id, userID string) (*models.UserScenario, error) { + log := trace.FromContext(ctx) + query := ` SELECT id, user_id, name, icon, description, prompt, greeting, language, created_at, updated_at FROM user_scenarios @@ -131,13 +144,18 @@ func (r *PostgresUserScenarioRepo) FindByIDAndUserID(ctx context.Context, id, us return nil, fmt.Errorf("user scenario not found or no permission") } if err != nil { + log.Errorw("find user scenario by id and user failed", "scenario_id", id, "user_id", userID, "error", err) return nil, fmt.Errorf("find user scenario: %w", err) } + + log.Debugw("user scenario found by id and user", "scenario_id", id, "user_id", userID) return &scenario, nil } // FindByUserID 查找用户的所有情景。 func (r *PostgresUserScenarioRepo) FindByUserID(ctx context.Context, userID string) ([]*models.UserScenario, error) { + log := trace.FromContext(ctx) + query := ` SELECT id, user_id, name, icon, description, prompt, greeting, language, created_at, updated_at FROM user_scenarios @@ -147,6 +165,7 @@ func (r *PostgresUserScenarioRepo) FindByUserID(ctx context.Context, userID stri rows, err := r.pool.Query(ctx, query, userID) if err != nil { + log.Errorw("find user scenarios failed", "user_id", userID, "error", err) return nil, fmt.Errorf("find user scenarios: %w", err) } defer rows.Close() @@ -167,19 +186,25 @@ func (r *PostgresUserScenarioRepo) FindByUserID(ctx context.Context, userID stri &s.UpdatedAt, ) if err != nil { + log.Errorw("scan user scenario row failed", "user_id", userID, "error", err) return nil, fmt.Errorf("scan user scenario: %w", err) } scenarios = append(scenarios, &s) } if err = rows.Err(); err != nil { + log.Errorw("iterate user scenarios failed", "user_id", userID, "error", err) return nil, fmt.Errorf("iterate user scenarios: %w", err) } + + log.Debugw("user scenarios found", "user_id", userID, "count", len(scenarios)) return scenarios, nil } // Update 更新用户情景。 func (r *PostgresUserScenarioRepo) Update(ctx context.Context, scenario *models.UserScenario) error { + log := trace.FromContext(ctx) + query := ` UPDATE user_scenarios SET name = $1, icon = $2, description = $3, prompt = $4, greeting = $5, language = $6, updated_at = $7 @@ -205,34 +230,47 @@ func (r *PostgresUserScenarioRepo) Update(ctx context.Context, scenario *models. return fmt.Errorf("user scenario not found or no permission") } if err != nil { + log.Errorw("update user scenario failed", "scenario_id", scenario.ID, "user_id", scenario.UserID, "error", err) return fmt.Errorf("update user scenario: %w", err) } + + log.Debugw("user scenario updated", "scenario_id", scenario.ID, "user_id", scenario.UserID) return nil } // Delete 删除用户情景。 func (r *PostgresUserScenarioRepo) Delete(ctx context.Context, id string) error { + log := trace.FromContext(ctx) + query := `DELETE FROM user_scenarios WHERE id = $1` result, err := r.pool.Exec(ctx, query, id) if err != nil { + log.Errorw("delete user scenario failed", "scenario_id", id, "error", err) return fmt.Errorf("delete user scenario: %w", err) } if result.RowsAffected() == 0 { return fmt.Errorf("user scenario not found") } + + log.Debugw("user scenario deleted", "scenario_id", id) return nil } // CountByUserID 统计用户的情景数量。 func (r *PostgresUserScenarioRepo) CountByUserID(ctx context.Context, userID string) (int, error) { + log := trace.FromContext(ctx) + query := `SELECT COUNT(*) FROM user_scenarios WHERE user_id = $1` var count int err := r.pool.QueryRow(ctx, query, userID).Scan(&count) if err != nil { + log.Errorw("count user scenarios failed", "user_id", userID, "error", err) return 0, fmt.Errorf("count user scenarios: %w", err) } + + log.Debugw("user scenarios counted", "user_id", userID, "count", count) return count, nil }