Implement batch processing for achievement saving with HTTP request handling

dbwriter.go

@@ -1,9 +1,27 @@
 package main
 
 import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
 	logger "git.site.quack-lab.dev/dave/cylogger"
 )
 
+const (
+	batchSize = 1000
+	timeout   = 5 * time.Second
+)
+
+var (
+	batchBuffer = make([]NSQMessage, 0, batchSize)
+	batchMutex  sync.Mutex
+)
+
 var whitelistedAchievements = map[string]bool{
 	"15":    true,
 	"958":   true,
@@ -65,19 +83,73 @@ var whitelistedAchievements = map[string]bool{
 	"12448": true,
 }
 
-func Save(message NSQMessage, db *DB) error {
+func Save(message NSQMessage) error {
 	_, ok := whitelistedAchievements[message.ID]
 	if !ok {
 		logger.Debug("Received message for non-whitelisted achievement %s", message.ID)
 		return nil
 	}
 
-	_, err := db.writeConn.Exec("INSERT OR IGNORE INTO achievements (name, id, date, completed) VALUES (?, ?, ?, ?)",
+	batchMutex.Lock()
-		message.Name, message.ID, message.Date, message.Completed)
+	batchBuffer = append(batchBuffer, message)
-	if err != nil {
+	currentBatchSize := len(batchBuffer)
-		logger.Error("Error inserting into database: %v", err)
+	batchMutex.Unlock()
-		return err
+
+	logger.Debug("Added achievement to batch. Current batch size: %d/%d", currentBatchSize, batchSize)
+
+	if currentBatchSize >= batchSize {
+		batchMutex.Lock()
+		batch := make([]NSQMessage, len(batchBuffer))
+		copy(batch, batchBuffer)
+		batchBuffer = batchBuffer[:0]
+		batchMutex.Unlock()
+
+		logger.Info("Batch size reached %d, sending batch to %s", len(batch), backendEndpoint)
+		return sendBatch(batch)
 	}
 
 	return nil
 }
+
+func sendBatch(batch []NSQMessage) error {
+	logger.Debug("Preparing to send batch of %d achievements", len(batch))
+
+	data, err := json.Marshal(batch)
+	if err != nil {
+		logger.Error("Failed to marshal batch: %v", err)
+		return fmt.Errorf("error marshaling batch: %v", err)
+	}
+	logger.Debug("Successfully marshaled batch to JSON, size: %d bytes", len(data))
+
+	client := &http.Client{
+		Timeout: timeout,
+	}
+
+	req, err := http.NewRequest("POST", backendEndpoint, bytes.NewBuffer(data))
+	if err != nil {
+		logger.Error("Failed to create HTTP request: %v", err)
+		return fmt.Errorf("error creating request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	logger.Debug("Created HTTP request to %s", backendEndpoint)
+
+	resp, err := client.Do(req)
+	if err != nil {
+		logger.Error("Failed to send HTTP request: %v", err)
+		return fmt.Errorf("error sending batch: %v", err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		logger.Error("Failed to read response body: %v", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		logger.Error("Received non-OK status code: %d, body: %s", resp.StatusCode, string(body))
+		return fmt.Errorf("unexpected status code: %d, body: %s", resp.StatusCode, string(body))
+	}
+
+	logger.Info("Successfully sent batch of %d achievements", len(batch))
+	return nil
+}

main.go

@@ -20,6 +20,7 @@ import (
 )
 
 const nsqEndpoint = "https://nsq.site.quack-lab.dev/pub?topic=wowspy"
+const backendEndpoint = "https://sniffer-be.site.quack-lab.dev/achievements"
 
 var debug *bool
 var nsqWorkers = 32
@@ -33,19 +34,9 @@ func main() {
 		logger.SetLevel(logger.LevelDebug) // Assuming LevelDebug is the correct constant for cylogger
 	}
 
-	db := DB{
-		path: "service/data/db.db",
-	}
-	err := db.Open()
-	if err != nil {
-		logger.Error("error opening database: %v", err)
-		return
-	}
-	defer db.Close()
-
 	logger.Info("Root: %q", *root)
 	cleanedRoot := strings.Replace(*root, "~", os.Getenv("HOME"), 1)
-	cleanedRoot, err = filepath.Abs(cleanedRoot)
+	cleanedRoot, err := filepath.Abs(cleanedRoot)
 	if err != nil {
 		logger.Error("error getting absolute path: %v", err)
 		return
@@ -75,33 +66,12 @@ func main() {
 	// We can save the achievements to the database while doing something else unrelated
 	go func() {
 		defer wg.Done()
-		saveAchievementsToDB(&db, achievements)
+		saveAchievements(achievements)
 	}()
-	loadUniquePlayers(&db, achievements)
+	// saveAchievementsToSourceFiles(luaStates, achievements)
-	saveAchievementsToSourceFiles(luaStates, achievements)
 	wg.Wait()
 }
 
-// loadUniquePlayers loads all unique players from the database and stores them in the achievements map
-// So that we can populate our "alreadySeen" with the database values too
-func loadUniquePlayers(db *DB, achievements *sync.Map) {
-	rows, err := db.readConn.Query("SELECT DISTINCT name FROM achievements")
-	if err != nil {
-		logger.Error("error querying unique players: %v", err)
-		return
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var playerName string
-		err = rows.Scan(&playerName)
-		if err != nil {
-			logger.Error("error scanning player name: %v", err)
-		}
-		achievements.Store(playerName, &[]NSQMessage{})
-	}
-	return
-}
-
 func saveAchievementsToSourceFiles(luaStates *sync.Map, achievements *sync.Map) {
 	wg := sync.WaitGroup{}
 	luaStates.Range(func(k, v any) bool {
@@ -202,14 +172,14 @@ func addAlreadySeen(strContent string, achievements *sync.Map) string {
 	return strings.Join(modifiedLines, "\n")
 }
 
-func saveAchievementsToDB(db *DB, achievements *sync.Map) {
+func saveAchievements(achievements *sync.Map) {
 	count := 0
 	achievements.Range(func(k, v any) bool {
 		playerName := k.(string)
 		playerAchievements := v.(*[]NSQMessage)
 		logger.Debug("Saving %d achievements for player %s", len(*playerAchievements), playerName)
 		for _, ach := range *playerAchievements {
-			Save(ach, db)
+			Save(ach)
 		}
 		count++
 		if count%1000 == 0 {

succ.sh

@@ -1 +1 @@
-go run . -root "C:/Users/Administrator/Seafile/Games-WoW/Ruski/WTF/"
+./main.exe -root "C:/Users/Administrator/Seafile/Games-WoW/Ruski/WTF/"