430 lines
10 KiB
Go
430 lines
10 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"datarush/internal/results/domain"
|
|
"datarush/internal/results/repository"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/lib/pq"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
const (
|
|
resultCachePrefix = "result:"
|
|
)
|
|
|
|
type ResultRepository struct {
|
|
db *sql.DB
|
|
redisClient *redis.Client
|
|
cacheEnabled bool
|
|
}
|
|
|
|
func NewResultRepository(db *sql.DB, redisClient *redis.Client, cacheEnabled bool) repository.ResultRepository {
|
|
return &ResultRepository{
|
|
db: db,
|
|
redisClient: redisClient,
|
|
cacheEnabled: cacheEnabled,
|
|
}
|
|
}
|
|
|
|
func (r *ResultRepository) cacheKey(id string) string {
|
|
return resultCachePrefix + id
|
|
}
|
|
|
|
func (r *ResultRepository) Create(ctx context.Context, result *domain.Result) error {
|
|
metadataJSON, err := json.Marshal(result.Metadata)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to marshal metadata: %w", err)
|
|
}
|
|
|
|
query := `INSERT INTO results (id, competition_id, user_id, score, rank, status, submission_id, metadata, created_at, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`
|
|
_, err = r.db.ExecContext(ctx, query,
|
|
result.ID,
|
|
result.CompetitionID,
|
|
result.UserID,
|
|
result.Score,
|
|
result.Rank,
|
|
result.Status,
|
|
result.SubmissionID,
|
|
metadataJSON,
|
|
result.CreatedAt,
|
|
result.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
if pqErr, ok := err.(*pq.Error); ok && pqErr.Code == "23505" {
|
|
return domain.ErrDuplicateResult
|
|
}
|
|
return err
|
|
}
|
|
|
|
if r.cacheEnabled {
|
|
data, err := json.Marshal(result)
|
|
if err == nil {
|
|
r.redisClient.Set(ctx, r.cacheKey(result.ID.String()), data, 10*time.Minute).Err()
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (r *ResultRepository) Get(ctx context.Context, id uuid.UUID) (*domain.Result, error) {
|
|
if r.cacheEnabled {
|
|
val, err := r.redisClient.Get(ctx, r.cacheKey(id.String())).Result()
|
|
if err == nil {
|
|
var result domain.Result
|
|
if json.Unmarshal([]byte(val), &result) == nil {
|
|
return &result, nil
|
|
}
|
|
}
|
|
}
|
|
|
|
query := `SELECT id, competition_id, user_id, score, rank, status, submission_id, metadata, created_at, updated_at
|
|
FROM results WHERE id = $1`
|
|
row := r.db.QueryRowContext(ctx, query, id)
|
|
|
|
var result domain.Result
|
|
var metadataJSON []byte
|
|
err := row.Scan(
|
|
&result.ID,
|
|
&result.CompetitionID,
|
|
&result.UserID,
|
|
&result.Score,
|
|
&result.Rank,
|
|
&result.Status,
|
|
&result.SubmissionID,
|
|
&metadataJSON,
|
|
&result.CreatedAt,
|
|
&result.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrResultNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
if len(metadataJSON) > 0 {
|
|
if err := json.Unmarshal(metadataJSON, &result.Metadata); err != nil {
|
|
return nil, fmt.Errorf("failed to unmarshal metadata: %w", err)
|
|
}
|
|
}
|
|
|
|
if r.cacheEnabled {
|
|
data, err := json.Marshal(&result)
|
|
if err == nil {
|
|
r.redisClient.Set(ctx, r.cacheKey(id.String()), data, 10*time.Minute).Err()
|
|
}
|
|
}
|
|
|
|
return &result, nil
|
|
}
|
|
|
|
func (r *ResultRepository) Update(ctx context.Context, result *domain.Result) error {
|
|
metadataJSON, err := json.Marshal(result.Metadata)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to marshal metadata: %w", err)
|
|
}
|
|
|
|
query := `UPDATE results SET
|
|
competition_id = $2, user_id = $3, score = $4, rank = $5, status = $6,
|
|
submission_id = $7, metadata = $8, updated_at = $9
|
|
WHERE id = $1`
|
|
res, err := r.db.ExecContext(ctx, query,
|
|
result.ID,
|
|
result.CompetitionID,
|
|
result.UserID,
|
|
result.Score,
|
|
result.Rank,
|
|
result.Status,
|
|
result.SubmissionID,
|
|
metadataJSON,
|
|
result.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rows, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if rows == 0 {
|
|
return domain.ErrResultNotFound
|
|
}
|
|
|
|
if r.cacheEnabled {
|
|
r.redisClient.Del(ctx, r.cacheKey(result.ID.String())).Err()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (r *ResultRepository) Delete(ctx context.Context, id uuid.UUID) error {
|
|
query := `DELETE FROM results WHERE id = $1`
|
|
res, err := r.db.ExecContext(ctx, query, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rows, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if rows == 0 {
|
|
return domain.ErrResultNotFound
|
|
}
|
|
|
|
if r.cacheEnabled {
|
|
r.redisClient.Del(ctx, r.cacheKey(id.String())).Err()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (r *ResultRepository) List(ctx context.Context, opts repository.ListResultsOptions) ([]domain.Result, int, error) {
|
|
var args []interface{}
|
|
var whereClauses []string
|
|
argId := 1
|
|
|
|
if opts.CompetitionID != nil {
|
|
whereClauses = append(whereClauses, fmt.Sprintf("competition_id = $%d", argId))
|
|
args = append(args, *opts.CompetitionID)
|
|
argId++
|
|
}
|
|
|
|
if opts.UserID != nil {
|
|
whereClauses = append(whereClauses, fmt.Sprintf("user_id = $%d", argId))
|
|
args = append(args, *opts.UserID)
|
|
argId++
|
|
}
|
|
|
|
if opts.Status != nil {
|
|
whereClauses = append(whereClauses, fmt.Sprintf("status = $%d", argId))
|
|
args = append(args, *opts.Status)
|
|
argId++
|
|
}
|
|
|
|
if opts.MinScore != nil {
|
|
whereClauses = append(whereClauses, fmt.Sprintf("score >= $%d", argId))
|
|
args = append(args, *opts.MinScore)
|
|
argId++
|
|
}
|
|
|
|
if opts.MaxScore != nil {
|
|
whereClauses = append(whereClauses, fmt.Sprintf("score <= $%d", argId))
|
|
args = append(args, *opts.MaxScore)
|
|
argId++
|
|
}
|
|
|
|
where := ""
|
|
if len(whereClauses) > 0 {
|
|
where = "WHERE " + strings.Join(whereClauses, " AND ")
|
|
}
|
|
|
|
countQuery := "SELECT COUNT(*) FROM results " + where
|
|
var total int
|
|
if err := r.db.QueryRowContext(ctx, countQuery, args...).Scan(&total); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
query := fmt.Sprintf(`SELECT id, competition_id, user_id, score, rank, status, submission_id, metadata, created_at, updated_at
|
|
FROM results %s ORDER BY score DESC, created_at ASC LIMIT $%d OFFSET $%d`, where, argId, argId+1)
|
|
args = append(args, opts.PageSize, opts.Page*opts.PageSize)
|
|
|
|
rows, err := r.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var results []domain.Result
|
|
for rows.Next() {
|
|
var result domain.Result
|
|
var metadataJSON []byte
|
|
err := rows.Scan(
|
|
&result.ID,
|
|
&result.CompetitionID,
|
|
&result.UserID,
|
|
&result.Score,
|
|
&result.Rank,
|
|
&result.Status,
|
|
&result.SubmissionID,
|
|
&metadataJSON,
|
|
&result.CreatedAt,
|
|
&result.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
if len(metadataJSON) > 0 {
|
|
if err := json.Unmarshal(metadataJSON, &result.Metadata); err != nil {
|
|
return nil, 0, fmt.Errorf("failed to unmarshal metadata: %w", err)
|
|
}
|
|
}
|
|
|
|
results = append(results, result)
|
|
}
|
|
|
|
return results, total, nil
|
|
}
|
|
|
|
func (r *ResultRepository) GetByCompetitionAndUser(ctx context.Context, competitionID uuid.UUID, userID uuid.UUID) (*domain.Result, error) {
|
|
query := `SELECT id, competition_id, user_id, score, rank, status, submission_id, metadata, created_at, updated_at
|
|
FROM results WHERE competition_id = $1 AND user_id = $2`
|
|
row := r.db.QueryRowContext(ctx, query, competitionID, userID)
|
|
|
|
var result domain.Result
|
|
var metadataJSON []byte
|
|
err := row.Scan(
|
|
&result.ID,
|
|
&result.CompetitionID,
|
|
&result.UserID,
|
|
&result.Score,
|
|
&result.Rank,
|
|
&result.Status,
|
|
&result.SubmissionID,
|
|
&metadataJSON,
|
|
&result.CreatedAt,
|
|
&result.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrResultNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
if len(metadataJSON) > 0 {
|
|
if err := json.Unmarshal(metadataJSON, &result.Metadata); err != nil {
|
|
return nil, fmt.Errorf("failed to unmarshal metadata: %w", err)
|
|
}
|
|
}
|
|
|
|
return &result, nil
|
|
}
|
|
|
|
func (r *ResultRepository) GetLeaderboard(ctx context.Context, competitionID uuid.UUID, limit int, offset int) ([]domain.Result, int, error) {
|
|
countQuery := `SELECT COUNT(*) FROM results WHERE competition_id = $1 AND status = $2`
|
|
var total int
|
|
if err := r.db.QueryRowContext(ctx, countQuery, competitionID, domain.ResultStatusCompleted).Scan(&total); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
query := `SELECT id, competition_id, user_id, score, rank, status, submission_id, metadata, created_at, updated_at
|
|
FROM results
|
|
WHERE competition_id = $1 AND status = $2
|
|
ORDER BY score DESC, created_at ASC
|
|
LIMIT $3 OFFSET $4`
|
|
|
|
rows, err := r.db.QueryContext(ctx, query, competitionID, domain.ResultStatusCompleted, limit, offset)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var results []domain.Result
|
|
for rows.Next() {
|
|
var result domain.Result
|
|
var metadataJSON []byte
|
|
err := rows.Scan(
|
|
&result.ID,
|
|
&result.CompetitionID,
|
|
&result.UserID,
|
|
&result.Score,
|
|
&result.Rank,
|
|
&result.Status,
|
|
&result.SubmissionID,
|
|
&metadataJSON,
|
|
&result.CreatedAt,
|
|
&result.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
if len(metadataJSON) > 0 {
|
|
if err := json.Unmarshal(metadataJSON, &result.Metadata); err != nil {
|
|
return nil, 0, fmt.Errorf("failed to unmarshal metadata: %w", err)
|
|
}
|
|
}
|
|
|
|
results = append(results, result)
|
|
}
|
|
|
|
return results, total, nil
|
|
}
|
|
|
|
func (r *ResultRepository) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.ResultStatus) (*domain.Result, error) {
|
|
query := `UPDATE results SET status = $1, updated_at = $2 WHERE id = $3`
|
|
now := time.Now()
|
|
res, err := r.db.ExecContext(ctx, query, status, now, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rows, err := res.RowsAffected()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if rows == 0 {
|
|
return nil, domain.ErrResultNotFound
|
|
}
|
|
|
|
if r.cacheEnabled {
|
|
r.redisClient.Del(ctx, r.cacheKey(id.String())).Err()
|
|
}
|
|
|
|
return r.Get(ctx, id)
|
|
}
|
|
|
|
func (r *ResultRepository) RecalculateRanks(ctx context.Context, competitionID uuid.UUID) error {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
query := `SELECT id FROM results
|
|
WHERE competition_id = $1 AND status = $2
|
|
ORDER BY score DESC, created_at ASC`
|
|
|
|
rows, err := tx.QueryContext(ctx, query, competitionID, domain.ResultStatusCompleted)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var ids []uuid.UUID
|
|
for rows.Next() {
|
|
var id uuid.UUID
|
|
if err := rows.Scan(&id); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
ids = append(ids, id)
|
|
}
|
|
rows.Close()
|
|
|
|
updateQuery := `UPDATE results SET rank = $1, updated_at = $2 WHERE id = $3`
|
|
now := time.Now()
|
|
for i, id := range ids {
|
|
rank := i + 1
|
|
if _, err := tx.ExecContext(ctx, updateQuery, rank, now, id); err != nil {
|
|
return err
|
|
}
|
|
|
|
if r.cacheEnabled {
|
|
r.redisClient.Del(ctx, r.cacheKey(id.String())).Err()
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|