ITQ/Datarush
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add redis cache to results

Browse Source
This commit is contained in:
timka
2025-12-17 18:38:54 +03:00
parent 7bf909652b
commit f5f6c7850c
4 changed files with 89 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.sum
-5
View File
@@ -138,13 +138,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
<<<<<<< HEAD
github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERSEP4=
github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
=======
github.com/redis/go-redis/v9 v9.17.2 h1:P2EGsA4qVIM3Pp+aPocCJ7DguDHhqrXNhVcEp4ViluI=
github.com/redis/go-redis/v9 v9.17.2/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
>>>>>>> origin/feature/task
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
internal/results/config/config.go
+11 -7
View File
@@ -11,18 +11,20 @@ import (
)
type Config struct {
GRPCPort int
GRPCEnableReflection bool
HTTPPort int
LogLevel string
DBHost string
DBPort int
GRPCPort int
GRPCEnableReflection bool
HTTPPort int
LogLevel string
DBHost string
DBPort int
DBUser string
DBPassword string
DBName string
RedisAddr string
UserServiceAddr string
SubmissionServiceAddr string
TaskServiceAddr string
CacheEnabled bool
}
func Load() (*Config, error) {
@@ -38,9 +40,11 @@ func Load() (*Config, error) {
DBUser: getEnv("POSTGRES_USERNAME", "postgres"),
DBPassword: getEnv("POSTGRES_PASSWORD", "postgres"),
DBName: getEnv("POSTGRES_DATABASE", "postgres"),
RedisAddr: getEnv("REDIS_ADDR", "localhost:6379"),
UserServiceAddr: getEnv("USER_GRPC_ADDR", "localhost:50051"),
SubmissionServiceAddr: getEnv("SUBMISSION_GRPC_ADDR", "localhost:50055"),
TaskServiceAddr: getEnv("TASK_GRPC_ADDR", "localhost:50056"),
CacheEnabled: mustGetBool("CACHE_ENABLED", false),
}, nil
}
@@ -77,4 +81,4 @@ func (c Config) BuildPostgresConnStr() string {
func (c Config) BuildPostgresDSN() string {
return fmt.Sprintf("postgresql://%s:%s@%s/%s?sslmode=disable",
c.DBUser, c.DBPassword, net.JoinHostPort(c.DBHost, strconv.Itoa(c.DBPort)), c.DBName)
}
}
internal/results/repository/postgres/result.go
+63 -3
View File
@@ -16,9 +16,15 @@ import (
)
const (
resultCachePrefix = "result:"
resultCachePrefix = "result:"
leaderboardCachePrefix = "leaderboard:"
)
type leaderboardCache struct {
Results []domain.Result `json:"results"`
TotalCount int `json:"total_count"`
}
type ResultRepository struct {
db *sql.DB
redisClient *redis.Client
@@ -37,6 +43,21 @@ func (r *ResultRepository) cacheKey(id string) string {
return resultCachePrefix + id
}
func (r *ResultRepository) leaderboardCacheKey(competitionID uuid.UUID, limit, offset int) string {
return fmt.Sprintf("%s%s:limit:%d:offset:%d", leaderboardCachePrefix, competitionID.String(), limit, offset)
}
func (r *ResultRepository) invalidateLeaderboardCache(ctx context.Context, competitionID uuid.UUID) {
if !r.cacheEnabled {
return
}
pattern := fmt.Sprintf("%s%s:*", leaderboardCachePrefix, competitionID.String())
keys, err := r.redisClient.Keys(ctx, pattern).Result()
if err == nil && len(keys) > 0 {
r.redisClient.Del(ctx, keys...).Err()
}
}
func (r *ResultRepository) Create(ctx context.Context, result *domain.Result) error {
metadataJSON, err := json.Marshal(result.Metadata)
if err != nil {
@@ -65,6 +86,7 @@ func (r *ResultRepository) Create(ctx context.Context, result *domain.Result) er
}
if r.cacheEnabled {
r.invalidateLeaderboardCache(ctx, result.CompetitionID)
data, err := json.Marshal(result)
if err == nil {
r.redisClient.Set(ctx, r.cacheKey(result.ID.String()), data, 10*time.Minute).Err()
@@ -160,6 +182,7 @@ func (r *ResultRepository) Update(ctx context.Context, result *domain.Result) er
}
if r.cacheEnabled {
r.invalidateLeaderboardCache(ctx, result.CompetitionID)
r.redisClient.Del(ctx, r.cacheKey(result.ID.String())).Err()
}
@@ -312,6 +335,16 @@ func (r *ResultRepository) GetByCompetitionAndUser(ctx context.Context, competit
}
func (r *ResultRepository) GetLeaderboard(ctx context.Context, competitionID uuid.UUID, limit int, offset int) ([]domain.Result, int, error) {
if r.cacheEnabled {
key := r.leaderboardCacheKey(competitionID, limit, offset)
val, err := r.redisClient.Get(ctx, key).Result()
if err == nil {
var cached leaderboardCache
if json.Unmarshal([]byte(val), &cached) == nil {
return cached.Results, cached.TotalCount, nil
}
}
}
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 {
@@ -359,6 +392,15 @@ func (r *ResultRepository) GetLeaderboard(ctx context.Context, competitionID uui
results = append(results, result)
}
if r.cacheEnabled {
key := r.leaderboardCacheKey(competitionID, limit, offset)
cacheData := leaderboardCache{Results: results, TotalCount: total}
data, err := json.Marshal(cacheData)
if err == nil {
r.redisClient.Set(ctx, key, data, 1*time.Minute).Err()
}
}
return results, total, nil
}
@@ -382,7 +424,16 @@ func (r *ResultRepository) UpdateStatus(ctx context.Context, id uuid.UUID, statu
r.redisClient.Del(ctx, r.cacheKey(id.String())).Err()
}
return r.Get(ctx, id)
result, err := r.Get(ctx, id)
if err != nil {
return nil, err
}
if r.cacheEnabled {
r.invalidateLeaderboardCache(ctx, result.CompetitionID)
}
return result, nil
}
func (r *ResultRepository) RecalculateRanks(ctx context.Context, competitionID uuid.UUID) error {
@@ -425,5 +476,14 @@ func (r *ResultRepository) RecalculateRanks(ctx context.Context, competitionID u
}
}
return tx.Commit()
err = tx.Commit()
if err != nil {
return err
}
if r.cacheEnabled {
r.invalidateLeaderboardCache(ctx, competitionID)
}
return nil
}
internal/results/server/server.go
+15 -2
View File
@@ -15,6 +15,7 @@ import (
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq"
"github.com/redis/go-redis/v9"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
)
@@ -30,6 +31,7 @@ type Server struct {
httpServer *http.Server
config *config.Config
db *sqlx.DB
redis *redis.Client
}
func New(cfg *config.Config) *Server {
@@ -45,6 +47,11 @@ func (s *Server) Start() error {
}
s.db = db
redisClient := redis.NewClient(&redis.Options{
Addr: s.config.RedisAddr,
})
s.redis = redisClient
go func() {
if err := s.startGRPCServer(); err != nil {
log.Fatalf("failed to start gRPC server: %v", err)
@@ -69,7 +76,7 @@ func (s *Server) startGRPCServer() error {
}
func (s *Server) registerGRPCServices() {
resultRepo := postgres.NewResultRepository(s.db.DB, nil, false)
resultRepo := postgres.NewResultRepository(s.db.DB, s.redis, s.config.CacheEnabled)
resultService := service.NewService(resultRepo)
resultHandler := grpcHandlers.NewResultsHandler(
resultService,
@@ -98,5 +105,11 @@ func (s *Server) Stop() {
}
}
if s.redis != nil {
if err := s.redis.Close(); err != nil {
log.Printf("failed to close redis connection: %v", err)
}
}
log.Println("results server stopped")
}
}