From 86f788893fa60673a164fa3dae77dfac2c84da6f Mon Sep 17 00:00:00 2001
From: rcourtman <courtmanr@gmail.com>
Date: Fri, 3 Oct 2025 12:56:27 +0000
Subject: [PATCH] Add concurrency checks for RateTracker and WebSocket hub

---
 .../ratetracker_concurrency_test.go           | 58 +++++++++++++++++++
 internal/websocket/hub_concurrency_test.go    | 55 ++++++++++++++++++
 2 files changed, 113 insertions(+)
 create mode 100644 internal/monitoring/ratetracker_concurrency_test.go
 create mode 100644 internal/websocket/hub_concurrency_test.go

diff --git a/internal/monitoring/ratetracker_concurrency_test.go b/internal/monitoring/ratetracker_concurrency_test.go
new file mode 100644
index 000000000..ca2ce9ab2
--- /dev/null
+++ b/internal/monitoring/ratetracker_concurrency_test.go
@@ -0,0 +1,58 @@
+package monitoring
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/types"
+)
+
+func TestRateTrackerConcurrentAccess(t *testing.T) {
+	rt := NewRateTracker()
+
+	const iterations = 1000
+
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			metrics := types.IOMetrics{
+				DiskRead:   int64(i * 100),
+				DiskWrite:  int64(i * 80),
+				NetworkIn:  int64(i * 60),
+				NetworkOut: int64(i * 40),
+				Timestamp:  time.Now().Add(time.Duration(i) * time.Millisecond),
+			}
+			rt.CalculateRates("guest-1", metrics)
+			time.Sleep(time.Microsecond)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			metrics := types.IOMetrics{
+				DiskRead:   int64(i * 50),
+				DiskWrite:  int64(i * 70),
+				NetworkIn:  int64(i * 30),
+				NetworkOut: int64(i * 20),
+				Timestamp:  time.Now().Add(time.Duration(i) * time.Millisecond),
+			}
+			rt.CalculateRates("guest-2", metrics)
+			time.Sleep(time.Microsecond)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations/10; i++ {
+			rt.Clear()
+			time.Sleep(5 * time.Microsecond)
+		}
+	}()
+
+	wg.Wait()
+}
diff --git a/internal/websocket/hub_concurrency_test.go b/internal/websocket/hub_concurrency_test.go
new file mode 100644
index 000000000..ecc6a520c
--- /dev/null
+++ b/internal/websocket/hub_concurrency_test.go
@@ -0,0 +1,55 @@
+package websocket
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestHubConcurrentClients(t *testing.T) {
+	hub := NewHub(nil)
+
+	// Start hub processing loop in background
+	go hub.Run()
+
+	const iterations = 200
+
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	// Simulate register/unregister from multiple goroutines
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			client := &Client{
+				hub:  hub,
+				send: make(chan []byte, 10),
+				id:   "client-register-" + string(rune(i)),
+			}
+			hub.register <- client
+			time.Sleep(time.Microsecond)
+			hub.unregister <- client
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			hub.BroadcastMessage(Message{Type: "test", Data: map[string]int{"iteration": i}})
+			time.Sleep(time.Microsecond)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			hub.SetAllowedOrigins([]string{"http://localhost", "http://example.com"})
+			time.Sleep(time.Microsecond)
+		}
+	}()
+
+	wg.Wait()
+
+	// Allow hub to process remaining messages
+	time.Sleep(10 * time.Millisecond)
+}