Claude/codebase review recommendations 3ixfnh

README.md

@@ -56,9 +56,10 @@ The platform provides real-time room-based chat with live analytics: message thr
 - **Per-connection rate limiting** — token-bucket throttle on inbound messages.
 - **Token auth** — optional HMAC-signed bearer tokens (enabled when `AUTH_SECRET` is set); falls back to a `userId` query param for local development.
 - **Configurable CORS / WebSocket origin allowlist.**
-- **Graceful degradation** — runs without DynamoDB (chat + live analytics still work, no persistence).
+- **Graceful degradation** — runs without DynamoDB (chat + live analytics still work, no persistence). Table init distinguishes an absent table from an unreachable one, so a throttle or transport blip is reported rather than silently retried as a create, and startup waits for the table to become `ACTIVE` before accepting writes.
 - **Voice Ops Briefings** — an anomaly detector watches the live metrics and narrates traffic events aloud in a control-room register via ElevenLabs TTS. Off by default; see below.
 - **Polished React frontend** — refined dark theme, design tokens, reusable UI primitives, avatars, virtualized message list, and a live metrics dashboard.
+- **Clean shutdown** — the hub's register/unregister/broadcast sends are guarded by its `done` channel, so once the event loop stops, callers return immediately instead of blocking on a channel nothing drains. Without this, every connection's read pump leaks on its deferred unregister.
 - **Comprehensive tests** — every backend package has a `_test.go`; the suite runs clean under `-race`.
 
 ## Quick Start
@@ -165,7 +166,9 @@ Recent message history for a room or a user. Optional `?limit=` (default 50, max
   "timestamp": "2026-06-16T10:30:00Z"
 }
 ```
-**Types:** `chat`, `system`, `join`, `leave`. **Validation:** username required (≤50 chars); chat content required (≤1000 chars); `messageId`/`timestamp`/`roomId` are server-authoritative.
+**Types:** `chat`, `system`, `join`, `leave`. **Validation:** username required (≤50 chars); chat content required (≤1000 chars).
+
+`messageId`, `timestamp`, `roomId`, `userId` and `username` are server-authoritative: the server overwrites all five on every inbound frame, regardless of what the client sent. This is load-bearing rather than cosmetic — `messageId` is the DynamoDB sort key and `timestamp` is the GSI sort key, so honouring a client-supplied value would let one connection overwrite another's stored row or pin itself to the head of room history.
 
 ## Voice Ops Briefings
 
@@ -186,6 +189,8 @@ An anomaly detector polls the live metrics every 5 seconds and narrates traffic
 
 **Detected conditions:** message-rate spike, p99 latency breach, connection surge, traffic dropout — each in a Yellow or Red severity band, debounced to at most one briefing per kind per 5 minutes.
 
+Rate-based conditions are evaluated against **complete** minutes: the newest slot of the 15-minute window is the minute currently in progress, and comparing that partial count against full-minute baselines reads as a dropout at the top of every minute. The detector ignores it. The cost is granularity — a genuine spike or dropout is visible once its minute closes, so up to ~60s later than the 5s poll interval alone would suggest.
+
 A sample script, as spoken:
 
 > Ops, analytics. Message-rate spike in the global room — four hundred messages per minute against a baseline of ninety. Monitoring.
@@ -284,7 +289,9 @@ npm run build            # tsc type-check + vite build
 npm run lint             # eslint
 ```
 
-> Note: `TestClient_PingPong` has a documented pre-existing race in the test harness — left as-is intentionally.
+> The full suite runs clean under `-race` in a single step, locally and in CI. The
+> historical `TestClient_PingPong` race was fixed by making the test's ping counter
+> an `atomic.Int64`.
 
 ## Deployment
 

backend/pkg/analytics/aggregator.go

@@ -19,12 +19,15 @@ func newWindow() *slidingWindow {
 	return &slidingWindow{lastTick: time.Now().Truncate(time.Minute)}
 }
 
+// increment counts one message into the current minute. The lock is held across
+// both the rotation check and the add: releasing it in between lets a concurrent
+// rotation advance idx and zero the slot after this call has chosen its index,
+// dropping the count into a slot that is immediately cleared.
 func (w *slidingWindow) increment() {
 	w.mu.Lock()
+	defer w.mu.Unlock()
 	w.advanceIfNeeded()
-	idx := w.idx
-	w.mu.Unlock()
-	w.slots[idx].Add(1)
+	w.slots[w.idx].Add(1)
 }
 
 // snapshot returns the counts for the last 15 minutes, oldest first.

backend/pkg/analytics/detector.go

@@ -179,19 +179,26 @@ func (d *Detector) Check(cur Metrics, now time.Time) []Anomaly {
 	return out
 }
 
-// rateWindow splits a MessagesPerMinute window into the current minute and the
-// mean of the preceding slots. ok is false when the window is too short to
-// compare against (fewer than 2 slots).
+// rateWindow splits a MessagesPerMinute window into the most recent COMPLETE
+// minute and the mean of the minutes before it. The final slot is the minute
+// currently in progress: it is a partial count that resets on every rotation, so
+// comparing it against full-minute baselines reports a dropout at the top of
+// every minute on a busy server. ok is false when the window is too short to hold
+// a complete minute plus a baseline.
+//
+// The cost of ignoring the partial slot is granularity: an event is only visible
+// once the minute containing it has closed, so detection lags by up to a minute.
 func rateWindow(perMin []int64) (current int64, baseline float64, ok bool) {
-	if len(perMin) < 2 {
+	if len(perMin) < 3 {
 		return 0, 0, false
 	}
-	current = perMin[len(perMin)-1]
+	complete := perMin[:len(perMin)-1] // drop the in-progress slot
+	current = complete[len(complete)-1]
 	var sum int64
-	for _, v := range perMin[:len(perMin)-1] {
+	for _, v := range complete[:len(complete)-1] {
 		sum += v
 	}
-	baseline = float64(sum) / float64(len(perMin)-1)
+	baseline = float64(sum) / float64(len(complete)-1)
 	return current, baseline, true
 }
 

backend/pkg/analytics/detector_test.go

@@ -15,14 +15,17 @@ func mkMetrics(perMin []int64, p99 float64, activeConns int64) Metrics {
 	}
 }
 
-// window builds a 15-slot MessagesPerMinute window: 14 baseline slots at `base`,
-// current minute at `cur`.
+// window builds a 15-slot MessagesPerMinute window: 13 baseline slots at `base`,
+// the most recent COMPLETE minute at `cur` in slot 13, and slot 14 left at zero.
+// Slot 14 is the minute currently in progress, which the detector ignores — see
+// rateWindow. Placing `cur` there instead would encode the bug this helper's
+// previous version was written against.
 func window(base, cur int64) []int64 {
 	w := make([]int64, 15)
-	for i := 0; i < 14; i++ {
+	for i := 0; i < 13; i++ {
 		w[i] = base
 	}
-	w[14] = cur
+	w[13] = cur
 	return w
 }
 
@@ -195,6 +198,38 @@ func TestDetector_TrafficDropout(t *testing.T) {
 	}
 }
 
+// A busy server rotating into a new minute must not report a dropout: the freshly
+// zeroed slot is the minute in progress, not silence. The runner polls every 5s
+// while the window rotates every 60s, so treating the partial slot as the current
+// rate fired a spurious traffic_dropout at the top of every minute — roughly 12
+// briefings an hour after debounce, each one an ElevenLabs charge.
+func TestDetector_NoDropoutAtMinuteRotation(t *testing.T) {
+	now := time.Now()
+	d := NewDetector(DefaultThresholds())
+
+	// Steady traffic. Slot 14 is the minute in progress, partway to ~200.
+	before := make([]int64, 15)
+	for i := 0; i < 14; i++ {
+		before[i] = 200
+	}
+	before[14] = 195
+
+	// The window rotates: every slot shifts left and the new slot 14 starts at 0.
+	after := make([]int64, 15)
+	for i := 0; i < 13; i++ {
+		after[i] = 200
+	}
+	after[13] = 195
+	after[14] = 0
+
+	if got := d.Check(mkMetrics(before, 5, 10), now); len(got) != 0 {
+		t.Fatalf("steady traffic should report nothing, got %v", got)
+	}
+	if got := d.Check(mkMetrics(after, 5, 10), now.Add(5*time.Second)); len(got) != 0 {
+		t.Fatalf("minute rotation on a busy server must not fire, got %v", got)
+	}
+}
+
 func TestDetector_Debounce(t *testing.T) {
 	now := time.Now()
 	th := DefaultThresholds()

backend/pkg/briefing/runner_test.go

@@ -53,12 +53,16 @@ func (c *capture) first() []byte {
 }
 
 // spikeMetrics is a snapshot that trips the rate-spike rule.
+// spikeMetrics builds a window the detector reads as a rate spike: a 90/min
+// baseline with 400 in slot 13, the most recent COMPLETE minute. Slot 14 is the
+// minute in progress and is deliberately left at zero — the detector ignores it,
+// so putting the spike there would produce a fixture that never fires.
 func spikeMetrics() analytics.Metrics {
 	perMin := make([]int64, 15)
-	for i := 0; i < 14; i++ {
+	for i := 0; i < 13; i++ {
 		perMin[i] = 90
 	}
-	perMin[14] = 400
+	perMin[13] = 400
 	return analytics.Metrics{MessagesPerMinute: perMin}
 }
 

backend/pkg/client/client.go

@@ -183,18 +183,14 @@ func (c *Client) readPump() {
 			continue
 		}
 
-		// Set client metadata (server overrides any client-supplied values)
+		// Server owns every identity and ordering field. MessageID is the DynamoDB sort
+		// key and Timestamp is the GSI sort key, so a client-supplied value would let one
+		// connection overwrite another's row or pin itself to the head of room history.
 		msg.UserID = c.userID
 		msg.Username = c.username
 		msg.RoomID = c.roomID
-
-		// Enrich with server-side fields the client doesn't set
-		if msg.MessageID == "" {
-			msg.MessageID = uuid.New().String()
-		}
-		if msg.Timestamp.IsZero() {
-			msg.Timestamp = time.Now().UTC()
-		}
+		msg.MessageID = uuid.New().String()
+		msg.Timestamp = time.Now().UTC()
 
 		// Validate
 		if err := msg.Validate(); err != nil {

backend/pkg/client/client_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/epw80/chat-analytics-platform/pkg/message"
+	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 )
 
@@ -429,13 +430,19 @@ func TestClient_MetadataOverride(t *testing.T) {
 	}
 	defer ws.Close()
 
-	// Send message with different userID/username (should be overridden)
+	// Send a message that forges every server-owned field. MessageID is the DynamoDB
+	// sort key and Timestamp is the GSI sort key: honouring either would let this
+	// connection overwrite another client's row or pin itself to the head of history.
+	forgedTimestamp := time.Date(2099, 1, 1, 0, 0, 0, 0, time.UTC)
 	msg := &message.Message{
-		Type:     message.TypeChat,
-		UserID:   "fakeUser",
-		Username: "FakeName",
-		Content:  "Hello",
-	}
+		Type:      message.TypeChat,
+		UserID:    "fakeUser",
+		Username:  "FakeName",
+		Content:   "Hello",
+		MessageID: "attacker-controlled",
+		Timestamp: forgedTimestamp,
+	}
+	sentAt := time.Now().UTC()
 	data, _ := msg.ToJSON()
 	err = ws.WriteMessage(websocket.TextMessage, data)
 	if err != nil {
@@ -462,6 +469,23 @@ func TestClient_MetadataOverride(t *testing.T) {
 	if broadcastMsg.Username != "ActualName" {
 		t.Errorf("expected username 'ActualName', got '%s'", broadcastMsg.Username)
 	}
+
+	// The forged sort keys must not survive: a replayed MessageID would overwrite the
+	// peer's row on PutItem, and a far-future Timestamp would pin this message to the
+	// head of the chronological GSI permanently.
+	if broadcastMsg.MessageID == "attacker-controlled" {
+		t.Error("client-supplied messageID was honoured; server must generate it")
+	}
+	if _, err := uuid.Parse(broadcastMsg.MessageID); err != nil {
+		t.Errorf("expected a server-generated UUID, got '%s': %v", broadcastMsg.MessageID, err)
+	}
+	if broadcastMsg.Timestamp.Equal(forgedTimestamp) {
+		t.Error("client-supplied timestamp was honoured; server must generate it")
+	}
+	if broadcastMsg.Timestamp.Before(sentAt) || broadcastMsg.Timestamp.After(time.Now().UTC()) {
+		t.Errorf("expected a server-side timestamp within the test window, got %s",
+			broadcastMsg.Timestamp)
+	}
 }
 
 // mockPersister implements the Persister interface for testing. Enqueue records

backend/pkg/hub/hub.go

@@ -153,23 +153,46 @@ func (h *Hub) Run() {
 	}
 }
 
-// Register adds a client to the hub
+// Register adds a client to the hub.
+//
+// The send is guarded by h.done: once Run has returned nothing drains these
+// channels, and an unguarded send would block the caller forever. After
+// shutdown this is a no-op rather than a Close — the shutdown branch in Run has
+// already closed every client it knew about, and Client.Close closes the send
+// channel, so closing a late registrant risks a double close. Its writePump
+// exits on the next failed ping write.
 func (h *Hub) Register(client any) {
-	if c, ok := client.(Client); ok {
-		h.register <- c
+	c, ok := client.(Client)
+	if !ok {
+		return
+	}
+	select {
+	case h.register <- c:
+	case <-h.done:
 	}
 }
 
-// Unregister removes a client from the hub
+// Unregister removes a client from the hub. Guarded by h.done for the reason
+// given on Register: readPump calls this from a defer, so an unguarded send
+// would leak every client goroutine at shutdown.
 func (h *Hub) Unregister(client any) {
-	if c, ok := client.(Client); ok {
-		h.unregister <- c
+	c, ok := client.(Client)
+	if !ok {
+		return
+	}
+	select {
+	case h.unregister <- c:
+	case <-h.done:
 	}
 }
 
 // Broadcast sends a message to all clients in the given room.
+// Guarded by h.done; see Register.
 func (h *Hub) Broadcast(roomID string, message []byte) {
-	h.broadcast <- broadcastRequest{roomID: roomID, data: message}
+	select {
+	case h.broadcast <- broadcastRequest{roomID: roomID, data: message}:
+	case <-h.done:
+	}
 }
 
 // BroadcastAll sends a message to every connected client across all rooms.

backend/pkg/hub/hub_test.go

@@ -286,6 +286,37 @@ func TestHub_Shutdown(t *testing.T) {
 	}
 }
 
+// After Run returns, nothing drains the register/unregister/broadcast channels.
+// Unguarded sends block forever, and because readPump calls Unregister from a
+// defer that leaks every client goroutine on shutdown. Every send method must
+// return promptly once the hub is down.
+func TestHub_SendsDoNotBlockAfterShutdown(t *testing.T) {
+	hub := newTestHub()
+	go hub.Run()
+
+	client := newMockClient("late")
+	hub.Register(client)
+	time.Sleep(10 * time.Millisecond)
+
+	hub.Shutdown()
+	time.Sleep(10 * time.Millisecond)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		hub.Register(newMockClient("after-shutdown"))
+		hub.Unregister(client)
+		hub.Broadcast("global", []byte("post-shutdown"))
+		hub.BroadcastAll([]byte("post-shutdown"))
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("hub send methods blocked after shutdown; client goroutines would leak")
+	}
+}
+
 func TestHub_UnregisterNonExistentClient(t *testing.T) {
 	hub := newTestHub()
 	go hub.Run()