Merge pull request #48 from Quaver/score-consumer

Add redis stream consumer for scores + clan rankings

cmd/database/migrations/12_clan_scores.down.sql

@@ -0,0 +1,7 @@
+BEGIN
+
+ ALTER TABLE clan_scores DROP COLUMN mode;
+ ALTER TABLE clan_scores DROP COLUMN timestamp;
+ DROP INDEX clan_scores_mode_index ON clan_scores;
+
+COMMIT;

cmd/database/migrations/12_clan_scores.up.sql

@@ -0,0 +1,12 @@
+BEGIN;
+
+ALTER TABLE clan_scores
+    ADD mode TINYINT NOT NULL;
+
+ALTER TABLE clan_scores
+    ADD timestamp BIGINT NOT NULL;
+
+CREATE INDEX clan_scores_mode_index
+    ON clan_scores (mode);
+
+COMMIT;

cmd/score-consumer/main.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"github.com/Quaver/api2/azure"
+	"github.com/Quaver/api2/config"
+	"github.com/Quaver/api2/db"
+	"github.com/Quaver/api2/webhooks"
+	"github.com/redis/go-redis/v9"
+	"github.com/sirupsen/logrus"
+	"gorm.io/gorm"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	configPath := flag.String("config", "../../config.json", "path to config file")
+	flag.Parse()
+
+	if err := config.Load(*configPath); err != nil {
+		logrus.Panic(err)
+	}
+
+	if !config.Instance.IsProduction {
+		logrus.SetLevel(logrus.DebugLevel)
+	}
+
+	db.ConnectMySQL()
+	db.InitializeRedis()
+	db.InitializeElasticSearch()
+	azure.InitializeClient()
+	webhooks.InitializeWebhooks()
+
+	go consumeScores()
+
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+	<-quit
+
+	logrus.Info("Exiting...")
+}
+
+func consumeScores() {
+	subject := "quaver:scores:stream"
+	consumersGroup := "api-score-consumer-group"
+
+	err := db.Redis.XGroupCreate(db.RedisCtx, subject, consumersGroup, "0").Err()
+
+	if err != nil {
+		logrus.Warn(err)
+	}
+
+	for {
+		entries, err := db.Redis.XReadGroup(db.RedisCtx, &redis.XReadGroupArgs{
+			Group:    consumersGroup,
+			Consumer: "api-score-consumer",
+			Streams:  []string{subject, ">"},
+			Count:    2,
+			Block:    0,
+			NoAck:    false,
+		}).Result()
+
+		if err != nil {
+			logrus.Fatal(err)
+		}
+
+		for i := 0; i < len(entries[0].Messages); i++ {
+			messageID := entries[0].Messages[i].ID
+			scoreStr := entries[0].Messages[i].Values["data"]
+
+			var score db.RedisScore
+
+			if err := json.Unmarshal([]byte(scoreStr.(string)), &score); err != nil {
+				logrus.Error(err)
+				break
+			}
+
+			logrus.Infof("New Score: %v (#%v) | Map #%v | Difficulty: %v | Score #%v | Rating: %v | Acc: %v%%",
+				score.User.Username, score.User.Id, score.Map.Id, score.Map.DifficultyRating,
+				score.Score.Id, score.Score.PerformanceRating, score.Score.Accuracy)
+
+			go func() {
+				if err := insertClanScore(&score); err != nil {
+					logrus.Error("Error inserting clan score: ", err)
+				}
+			}()
+
+			db.Redis.XAck(db.RedisCtx, subject, consumersGroup, messageID)
+		}
+	}
+}
+
+// Handles the insertion of a new clan score
+func insertClanScore(score *db.RedisScore) error {
+	if !score.Map.ClanRanked || score.User.ClanId <= 0 {
+		return nil
+	}
+
+	existingScore, err := db.GetClanScore(score.Map.MD5, score.User.ClanId)
+
+	if err != nil && err != gorm.ErrRecordNotFound {
+		return err
+	}
+
+	newScore, err := db.CalculateClanScore(score.Map.MD5, score.User.ClanId, score.Map.GameMode)
+
+	if err != nil {
+		return err
+	}
+
+	// Make sure the id is the same on the newly calculated score, so it can be upserted properly.
+	if existingScore != nil {
+		newScore.Id = existingScore.Id
+	}
+
+	if err := db.SQL.Save(&newScore).Error; err != nil {
+		return err
+	}
+
+	if err := db.RecalculateClanStats(score.User.ClanId, score.Map.GameMode, score); err != nil {
+		return err
+	}
+
+	return nil
+}

db/clan_scores.go

@@ -1,13 +1,80 @@
 package db
 
+import (
+	"github.com/Quaver/api2/enums"
+	"time"
+)
+
 type ClanScore struct {
-	Id              int     `gorm:"column:id; PRIMARY_KEY"`
-	ClanId          int     `gorm:"column:clan_id"`
-	MapMD5          string  `gorm:"column:map_md5"`
-	OverallRating   float64 `gorm:"column:overall_rating"`
-	OverallAccuracy float64 `gorm:"column:overall_accuracy"`
+	Id              int            `gorm:"column:id; PRIMARY_KEY"`
+	ClanId          int            `gorm:"column:clan_id"`
+	MapMD5          string         `gorm:"column:map_md5"`
+	Mode            enums.GameMode `gorm:"column:mode"`
+	OverallRating   float64        `gorm:"column:overall_rating"`
+	OverallAccuracy float64        `gorm:"column:overall_accuracy"`
+	Timestamp       int64          `gorm:"column:timestamp"`
 }
 
 func (*ClanScore) TableName() string {
 	return "clan_scores"
 }
+
+// ToScore Converts a clan score to a traditional score object, so we can reuse functions to calculate rating/acc
+func (cs *ClanScore) ToScore() *Score {
+	return &Score{
+		ClanId:            &cs.ClanId,
+		MapMD5:            cs.MapMD5,
+		PerformanceRating: cs.OverallRating,
+		Accuracy:          cs.OverallAccuracy,
+	}
+}
+
+// GetClanScore Retrieves an existing clan score
+func GetClanScore(md5 string, clanId int) (*ClanScore, error) {
+	var score ClanScore
+
+	result := SQL.
+		Where("map_md5 = ? AND clan_id = ?", md5, clanId).
+		First(&score)
+
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return &score, nil
+}
+
+// GetClanScoresForMode Retrieves all clan scores for a given mode
+func GetClanScoresForMode(clanId int, mode enums.GameMode) ([]*ClanScore, error) {
+	clanScores := make([]*ClanScore, 0)
+
+	result := SQL.
+		Where("clan_id = ? AND mode = ?", clanId, mode).
+		Find(&clanScores)
+
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return clanScores, nil
+}
+
+// CalculateClanScore Calculates a clan score for a given map
+func CalculateClanScore(md5 string, clanId int, mode enums.GameMode) (*ClanScore, error) {
+	scores, err := GetClanPlayerScoresOnMap(md5, clanId)
+
+	if err != nil {
+		return nil, err
+	}
+
+	score := &ClanScore{
+		ClanId:          clanId,
+		MapMD5:          md5,
+		Mode:            mode,
+		OverallRating:   CalculateOverallRating(scores),
+		OverallAccuracy: CalculateOverallAccuracy(scores),
+		Timestamp:       time.Now().UnixMilli(),
+	}
+
+	return score, nil
+}

db/clan_stats.go

@@ -1,18 +1,116 @@
 package db
 
+import (
+	"github.com/Quaver/api2/enums"
+)
+
 type ClanStats struct {
-	ClanId                   int     `gorm:"column:clan_id" json:"clan_id"`
-	Mode                     int     `gorm:"column:mode" json:"mode"`
-	OverallAccuracy          float64 `gorm:"column:overall_accuracy" json:"overall_accuracy"`
-	OverallPerformanceRating float64 `gorm:"column:overall_performance_rating" json:"overall_performance_rating"`
-	TotalMarv                int     `gorm:"column:total_marv" json:"total_marv"`
-	TotalPerf                int     `gorm:"column:total_perf" json:"total_perf"`
-	TotalGreat               int     `gorm:"column:total_great" json:"total_great"`
-	TotalGood                int     `gorm:"column:total_good" json:"total_good"`
-	TotalOkay                int     `gorm:"column:total_okay" json:"total_okay"`
-	TotalMiss                int     `gorm:"column:total_miss" json:"total_miss"`
+	ClanId                   int            `gorm:"column:clan_id" json:"clan_id"`
+	Mode                     enums.GameMode `gorm:"column:mode" json:"mode"`
+	OverallAccuracy          float64        `gorm:"column:overall_accuracy" json:"overall_accuracy"`
+	OverallPerformanceRating float64        `gorm:"column:overall_performance_rating" json:"overall_performance_rating"`
+	TotalMarv                int            `gorm:"column:total_marv" json:"total_marv"`
+	TotalPerf                int            `gorm:"column:total_perf" json:"total_perf"`
+	TotalGreat               int            `gorm:"column:total_great" json:"total_great"`
+	TotalGood                int            `gorm:"column:total_good" json:"total_good"`
+	TotalOkay                int            `gorm:"column:total_okay" json:"total_okay"`
+	TotalMiss                int            `gorm:"column:total_miss" json:"total_miss"`
 }
 
 func (*ClanStats) TableName() string {
 	return "clan_stats"
 }
+
+func (cs *ClanStats) Save() error {
+	return SQL.Model(&ClanStats{}).
+		Where("clan_id = ? AND mode = ?", cs.ClanId, cs.Mode).
+		Update("overall_accuracy", cs.OverallAccuracy).
+		Update("overall_performance_rating", cs.OverallPerformanceRating).
+		Update("total_marv", cs.TotalMarv).
+		Update("total_perf", cs.TotalPerf).
+		Update("total_great", cs.TotalGreat).
+		Update("total_good", cs.TotalGood).
+		Update("total_okay", cs.TotalOkay).
+		Update("total_miss", cs.TotalMiss).Error
+}
+
+// GetClanStatsByMode Retrieves clan stats by its game mode
+func GetClanStatsByMode(id int, mode enums.GameMode) (*ClanStats, error) {
+	var stats ClanStats
+
+	result := SQL.
+		Where("clan_id = ? AND mode = ?", id, mode).
+		First(&stats)
+
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return &stats, nil
+}
+
+// PerformFullClanRecalculation Recalculates all of a clan's scores + stats
+func PerformFullClanRecalculation(clanId int) error {
+	for i := 1; i <= 2; i++ {
+		clanScores, err := GetClanScoresForMode(clanId, enums.GameMode(i))
+
+		if err != nil {
+			return err
+		}
+
+		for _, clanScore := range clanScores {
+			newScore, err := CalculateClanScore(clanScore.MapMD5, clanId, clanScore.Mode)
+
+			if err != nil {
+				return err
+			}
+
+			newScore.Id = clanScore.Id
+
+			if err := SQL.Save(&newScore).Error; err != nil {
+				return err
+			}
+		}
+
+		if err := RecalculateClanStats(clanId, enums.GameMode(i)); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// RecalculateClanStats Recalculates a clan stats for a given mode.
+func RecalculateClanStats(id int, mode enums.GameMode, newScore ...*RedisScore) error {
+	stats, err := GetClanStatsByMode(id, mode)
+
+	if err != nil {
+		return err
+	}
+
+	if len(newScore) > 0 {
+		stats.TotalMarv += newScore[0].Score.CountMarvelous
+		stats.TotalPerf += newScore[0].Score.CountPerfect
+		stats.TotalGreat += newScore[0].Score.CountGreat
+		stats.TotalGood += newScore[0].Score.CountGood
+		stats.TotalOkay += newScore[0].Score.CountOkay
+		stats.TotalMiss += newScore[0].Score.CountMiss
+	}
+
+	clanScores, err := GetClanScoresForMode(id, mode)
+
+	if err != nil {
+		return err
+	}
+
+	convertedScores := make([]*Score, 0)
+
+	for _, clanScore := range clanScores {
+		convertedScores = append(convertedScores, clanScore.ToScore())
+	}
+
+	stats.OverallPerformanceRating = CalculateOverallRating(convertedScores)
+	stats.OverallAccuracy = CalculateOverallAccuracy(convertedScores)
+
+	return stats.Save()
+}

db/clans.go

@@ -54,7 +54,7 @@ func (clan *Clan) Insert() error {
 		}
 
 		for i := 1; i <= 2; i++ {
-			if err := tx.Create(&ClanStats{ClanId: clan.Id, Mode: i}).Error; err != nil {
+			if err := tx.Create(&ClanStats{ClanId: clan.Id, Mode: enums.GameMode(i)}).Error; err != nil {
 				return err
 			}
 		}