Updates

internal/geomap/geomap_test.go

@@ -76,6 +76,39 @@ func TestStateRankedAndPulse(t *testing.T) {
 	}
 }
 
+func TestSetStatsMaxMergesAllTime(t *testing.T) {
+	s := NewState()
+	s.SetStats(map[string]int{"CN": 100, "US": 10}, map[string]int{"CN": 5}, nil)
+	s.BumpGeo("CN") // all=101, hour=6
+	// Sparser server snapshot must not wipe optimistic all-time bump.
+	s.SetStats(map[string]int{"CN": 100, "BR": 3}, map[string]int{"US": 2}, nil)
+	if s.CountFor("CN") != 101 {
+		s.SetRange(RangeAll)
+		if s.CountFor("CN") != 101 {
+			t.Fatalf("all-time CN want 101 got %d", s.CountFor("CN"))
+		}
+	}
+	s.SetRange(RangeAll)
+	if s.CountFor("BR") != 3 {
+		t.Fatalf("BR from server: %d", s.CountFor("BR"))
+	}
+	// Hour replaced by server window
+	s.SetRange(RangeHour)
+	if s.CountFor("US") != 2 || s.CountFor("CN") != 0 {
+		t.Fatalf("hour replace: US=%d CN=%d", s.CountFor("US"), s.CountFor("CN"))
+	}
+}
+
+func TestMergeFeedCountsEnrichesAll(t *testing.T) {
+	s := NewState()
+	s.SetStats(map[string]int{"CN": 5}, map[string]int{"CN": 1}, nil)
+	s.MergeFeedCounts(map[string]int{"CN": 12, "US": 4}, map[string]int{"US": 2})
+	s.SetRange(RangeAll)
+	if s.CountFor("CN") != 12 || s.CountFor("US") != 4 {
+		t.Fatalf("all after feed merge CN=%d US=%d", s.CountFor("CN"), s.CountFor("US"))
+	}
+}
+
 func TestRenderProducesOutput(t *testing.T) {
 	s := NewState()
 	s.SetStats(map[string]int{"CN": 100, "US": 40}, nil, map[string]int{"US": 1})

internal/geomap/state.go

@@ -164,17 +164,45 @@ func (s *State) Hover() string {
 	return s.hover
 }
 
-// SetStats replaces aggregate geo maps from viz.Stats.
+// SetStats merges aggregate geo maps from viz.Stats into local state.
+//
+// All-time counts are max-merged so live SSE bumps and feed-derived tallies
+// are never wiped by a sparser server snapshot. Last-hour counts are replaced
+// when the server provides a map (sliding window), otherwise left intact.
+// Peers-by-geo is replaced when non-nil.
 func (s *State) SetStats(all, hour, peers map[string]int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.allCounts = cloneCounts(all)
-	s.hourCounts = cloneCounts(hour)
-	s.peersByGeo = cloneCounts(peers)
+	s.allCounts = maxMergeCounts(s.allCounts, cloneCounts(all))
+	if hour != nil {
+		// Sliding window: trust server map when present (including empty).
+		s.hourCounts = cloneCounts(hour)
+	}
+	if peers != nil {
+		s.peersByGeo = cloneCounts(peers)
+	}
+}
+
+// MergeFeedCounts folds in-memory feed tallies into map counters (max-merge).
+// Used so the ALL range reflects attacks seen in the live buffer, not only
+// the server geoBreakdown window.
+func (s *State) MergeFeedCounts(all, hour map[string]int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.allCounts = maxMergeCounts(s.allCounts, cloneCounts(all))
+	// Hour: max-merge so live buffer + optimistic bumps survive until the next
+	// authoritative hour snapshot replaces hourCounts via SetStats.
+	s.hourCounts = maxMergeCounts(s.hourCounts, cloneCounts(hour))
 }
 
 // BumpGeo increments live counters when an attack arrives (optimistic).
+// Live SSE events are treated as "now" for the hour window.
 func (s *State) BumpGeo(code string) {
+	s.BumpGeoAt(code, true)
+}
+
+// BumpGeoAt increments all-time always; hour only when inHour is true.
+func (s *State) BumpGeoAt(code string, inHour bool) {
 	code = NormalizeCode(code)
 	if code == "" || code == UnknownGeo {
 		return
@@ -182,7 +210,25 @@ func (s *State) BumpGeo(code string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.allCounts[code]++
-	s.hourCounts[code]++
+	if inHour {
+		s.hourCounts[code]++
+	}
+}
+
+func maxMergeCounts(a, b map[string]int) map[string]int {
+	if a == nil && b == nil {
+		return map[string]int{}
+	}
+	out := make(map[string]int, len(a)+len(b))
+	for k, v := range a {
+		out[k] = v
+	}
+	for k, v := range b {
+		if cur, ok := out[k]; !ok || v > cur {
+			out[k] = v
+		}
+	}
+	return out
 }
 
 // TriggerPulse adds a visual attack flash.

internal/ui/map_tab.go

@@ -39,8 +39,21 @@ func (m *Model) syncMapFromStats() {
 				peers[g]++
 			}
 		}
+		if len(peers) == 0 {
+			peers = nil // don't wipe existing peer geo via SetStats
+		}
 	}
+	// Max-merge server KPI maps (all-time never shrinks; hour replaced when non-nil).
 	m.geoMap.SetStats(all, hour, peers)
+
+	// Fold every geo-tagged event in the live ring buffer so ALL reflects
+	// attacks we've actually processed, not only the server's geoBreakdown
+	// window (which can be much smaller than totalAttacks).
+	if m.store != nil {
+		nowMS := time.Now().UnixMilli()
+		feedAll, feedHour := viz.GeoCountsFromFeed(m.store.Items(), nowMS)
+		m.geoMap.MergeFeedCounts(feedAll, feedHour)
+	}
 }
 
 func (m *Model) mapTickCmd() tea.Cmd {

internal/viz/stats_live.go

@@ -132,16 +132,32 @@ func MergeAuthoritativeStats(prev *Stats, next Stats, opts MergeStatsOpts) *Stat
 	out.AttacksLastHour = pick(prev.AttacksLastHour, next.AttacksLastHour)
 
 	if next.ServiceBreakdown != nil {
-		out.ServiceBreakdown = cloneIntMap(next.ServiceBreakdown)
+		if opts.Absolute {
+			out.ServiceBreakdown = cloneIntMap(next.ServiceBreakdown)
+		} else {
+			out.ServiceBreakdown = MaxMergeIntMap(prev.ServiceBreakdown, next.ServiceBreakdown)
+		}
 	}
+	// All-time geo is monotonic: never drop optimistic live bumps or richer
+	// local tallies when the server sends a sparser snapshot.
 	if next.GeoBreakdown != nil {
-		out.GeoBreakdown = cloneIntMap(next.GeoBreakdown)
+		out.GeoBreakdown = MaxMergeIntMap(prev.GeoBreakdown, next.GeoBreakdown)
 	}
+	// Last-hour geo is a sliding window — take absolute server values when
+	// Absolute, otherwise max-merge so live SSE bumps survive between ticks.
 	if next.GeoBreakdownLastHour != nil {
-		out.GeoBreakdownLastHour = cloneIntMap(next.GeoBreakdownLastHour)
+		if opts.Absolute {
+			out.GeoBreakdownLastHour = cloneIntMap(next.GeoBreakdownLastHour)
+		} else {
+			out.GeoBreakdownLastHour = MaxMergeIntMap(prev.GeoBreakdownLastHour, next.GeoBreakdownLastHour)
+		}
 	}
 	if next.PeersByGeo != nil {
-		out.PeersByGeo = cloneIntMap(next.PeersByGeo)
+		if opts.Absolute {
+			out.PeersByGeo = cloneIntMap(next.PeersByGeo)
+		} else {
+			out.PeersByGeo = MaxMergeIntMap(prev.PeersByGeo, next.PeersByGeo)
+		}
 	}
 	if next.RecentBlocksLastHour != 0 || opts.Absolute {
 		out.RecentBlocksLastHour = next.RecentBlocksLastHour
@@ -149,6 +165,75 @@ func MergeAuthoritativeStats(prev *Stats, next Stats, opts MergeStatsOpts) *Stat
 	return &out
 }
 
+// MaxMergeIntMap returns a map where each key is max(a[k], b[k]).
+// Nil maps are treated as empty. Keys are preserved from both sides.
+func MaxMergeIntMap(a, b map[string]int) map[string]int {
+	if a == nil && b == nil {
+		return map[string]int{}
+	}
+	out := cloneIntMap(a)
+	for k, v := range b {
+		if cur, ok := out[k]; !ok || v > cur {
+			out[k] = v
+		}
+	}
+	return out
+}
+
+// GeoCountsFromFeed tallies geos from in-memory feed items.
+// all gets every geo-tagged attack/block/moderation; hour only events with
+// Timestamp within the last hour of nowMS.
+func GeoCountsFromFeed(items []FeedItem, nowMS int64) (all, hour map[string]int) {
+	all = map[string]int{}
+	hour = map[string]int{}
+	oneHourAgo := nowMS - hourMS
+	for _, it := range items {
+		switch it.Type {
+		case FeedAttack, FeedBlock, FeedModeration:
+		default:
+			continue
+		}
+		geo := normalizeGeoKey(it.Geo)
+		if geo == "" {
+			continue
+		}
+		all[geo]++
+		ts := it.Timestamp
+		if ts <= 0 {
+			ts = nowMS
+		}
+		if ts >= oneHourAgo {
+			hour[geo]++
+		}
+	}
+	return all, hour
+}
+
+func normalizeGeoKey(geo string) string {
+	if geo == "" {
+		return ""
+	}
+	// Uppercase ISO-ish codes without importing geomap (avoid cycle).
+	b := make([]byte, 0, len(geo))
+	for i := 0; i < len(geo); i++ {
+		c := geo[i]
+		if c >= 'a' && c <= 'z' {
+			c -= 'a' - 'A'
+		}
+		if (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') {
+			b = append(b, c)
+		}
+	}
+	s := string(b)
+	if s == "UK" {
+		return "GB"
+	}
+	if s == "UNKNOWN" || s == "XX" || s == "ZZ" {
+		return ""
+	}
+	return s
+}
+
 // BumpPeerOnAttack updates per-peer hour/service counters on the live roster.
 func BumpPeerOnAttack(net *Network, a AttackEvent, nowMS int64) *Network {
 	if net == nil || a.PeerID == "" {

internal/viz/stats_live_test.go

@@ -67,21 +67,72 @@ func TestBumpConnectedPeers(t *testing.T) {
 }
 
 func TestMergeAuthoritativeStats(t *testing.T) {
-	prev := &viz.Stats{TotalAttacks: 100, ConnectedPeers: 5, BlockedIPs: 10, AttacksLastHour: 20}
-	next := viz.Stats{TotalAttacks: 90, ConnectedPeers: 4, BlockedIPs: 12, AttacksLastHour: 15, ServiceBreakdown: map[string]int{"SSH": 1}}
+	prev := &viz.Stats{
+		TotalAttacks: 100, ConnectedPeers: 5, BlockedIPs: 10, AttacksLastHour: 20,
+		GeoBreakdown:         map[string]int{"CN": 50, "US": 10},
+		GeoBreakdownLastHour: map[string]int{"CN": 5},
+		ServiceBreakdown:     map[string]int{"SSH": 9},
+	}
+	next := viz.Stats{
+		TotalAttacks: 90, ConnectedPeers: 4, BlockedIPs: 12, AttacksLastHour: 15,
+		ServiceBreakdown:     map[string]int{"SSH": 1, "HTTP": 2},
+		GeoBreakdown:         map[string]int{"CN": 40, "BR": 3}, // sparser CN
+		GeoBreakdownLastHour: map[string]int{"US": 2},
+	}
 
 	merged := viz.MergeAuthoritativeStats(prev, next, viz.MergeStatsOpts{})
 	if merged.TotalAttacks != 100 || merged.ConnectedPeers != 5 || merged.BlockedIPs != 12 || merged.AttacksLastHour != 20 {
 		t.Fatalf("non-absolute merge: %+v", merged)
 	}
-	if merged.ServiceBreakdown["SSH"] != 1 {
-		t.Fatalf("breakdown replaced: %+v", merged.ServiceBreakdown)
+	// Service: max-merge non-absolute
+	if merged.ServiceBreakdown["SSH"] != 9 || merged.ServiceBreakdown["HTTP"] != 2 {
+		t.Fatalf("service max-merge: %+v", merged.ServiceBreakdown)
+	}
+	// All-time geo never drops local CN=50
+	if merged.GeoBreakdown["CN"] != 50 || merged.GeoBreakdown["BR"] != 3 || merged.GeoBreakdown["US"] != 10 {
+		t.Fatalf("geo all max-merge: %+v", merged.GeoBreakdown)
+	}
+	if merged.GeoBreakdownLastHour["CN"] != 5 || merged.GeoBreakdownLastHour["US"] != 2 {
+		t.Fatalf("geo hour max-merge: %+v", merged.GeoBreakdownLastHour)
 	}
 
 	abs := viz.MergeAuthoritativeStats(prev, next, viz.MergeStatsOpts{Absolute: true})
 	if abs.TotalAttacks != 90 || abs.ConnectedPeers != 4 || abs.AttacksLastHour != 15 {
 		t.Fatalf("absolute merge: %+v", abs)
 	}
+	// Absolute still max-merges all-time geo (monotonic)
+	if abs.GeoBreakdown["CN"] != 50 || abs.GeoBreakdown["BR"] != 3 {
+		t.Fatalf("absolute geo all should max-merge: %+v", abs.GeoBreakdown)
+	}
+	// Absolute replaces hour window
+	if abs.GeoBreakdownLastHour["US"] != 2 || abs.GeoBreakdownLastHour["CN"] != 0 {
+		t.Fatalf("absolute geo hour replace: %+v", abs.GeoBreakdownLastHour)
+	}
+}
+
+func TestGeoCountsFromFeed(t *testing.T) {
+	now := int64(1_000_000_000)
+	items := []viz.FeedItem{
+		{Type: viz.FeedAttack, Geo: "cn", Timestamp: now},
+		{Type: viz.FeedAttack, Geo: "CN", Timestamp: now},
+		{Type: viz.FeedAttack, Geo: "US", Timestamp: now - 2*60*60*1000},
+		{Type: viz.FeedPeer, Geo: "DE", Timestamp: now}, // ignored
+		{Type: viz.FeedModeration, Geo: "br", Timestamp: now - 30*60*1000},
+	}
+	all, hour := viz.GeoCountsFromFeed(items, now)
+	if all["CN"] != 2 || all["US"] != 1 || all["BR"] != 1 {
+		t.Fatalf("all: %+v", all)
+	}
+	if hour["CN"] != 2 || hour["BR"] != 1 || hour["US"] != 0 {
+		t.Fatalf("hour: %+v", hour)
+	}
+}
+
+func TestMaxMergeIntMap(t *testing.T) {
+	got := viz.MaxMergeIntMap(map[string]int{"A": 5, "B": 1}, map[string]int{"A": 3, "C": 9})
+	if got["A"] != 5 || got["B"] != 1 || got["C"] != 9 {
+		t.Fatalf("%+v", got)
+	}
 }
 
 func TestBumpPeerOnAttack(t *testing.T) {