node: High usage monitor (#4506)

### Description

Implemented abuse detection for event RPCs:

- Introduced a configurable CallRateMonitor that tracks per-user call
rates across sliding windows via ring buffers, plus helper wiring to
instantiate it from config.AbuseDetection.
- Hooked AddEvent, AddMediaEvent, and CreateMediaStream so every
successful call records the creator’s address under the appropriate call
type
- Exposed recent usage on the /status endpoint via a new field,
detailing which accounts exceeded which thresholds over 1-minute and
30-minute windows by default.

### Checklist

- [ ] Tests added where required
- [ ] Documentation updated where applicable
- [ ] Changes adhere to the repository's contribution guidelines

Author: jontra

core/config/config.go

@@ -53,6 +53,18 @@ func GetDefaultConfig() *Config {
 		Metrics: MetricsConfig{
 			Enabled: true,
 		},
+		HighUsageDetection: HighUsageDetectionConfig{
+			Enabled:    false,
+			MaxResults: 10,
+			Thresholds: HighUsageThresholdFields{
+				ThresholdAddEventWindow:          time.Minute,
+				ThresholdAddEventCount:           100,
+				ThresholdAddMediaEventWindow:     time.Minute,
+				ThresholdAddMediaEventCount:      50,
+				ThresholdCreateMediaStreamWindow: time.Minute,
+				ThresholdCreateMediaStreamCount:  10,
+			},
+		},
 		// TODO: Network: NetworkConfig{},
 		StandByOnStart:    true,
 		StandByPollPeriod: 500 * time.Millisecond,
@@ -122,6 +134,7 @@ type Config struct {
 	// Metrics
 	Metrics             MetricsConfig
 	PerformanceTracking PerformanceTrackingConfig
+	HighUsageDetection  HighUsageDetectionConfig
 
 	// Scrubbing
 	Scrubbing ScrubbingConfig
@@ -602,6 +615,60 @@ type MetricsConfig struct {
 	Interface string
 }
 
+type HighUsageDetectionConfig struct {
+	// Enabled toggles the high-usage detection tracker logic.
+	Enabled bool
+
+	// MaxResults limits the number of high-usage accounts exposed via /status.
+	MaxResults int
+
+	// Thresholds captures explicit per-call-type threshold definitions.
+	Thresholds HighUsageThresholdFields
+}
+
+// HighUsageThresholds flattens the configured threshold_* fields into a standard
+// map keyed by call type.
+func (cfg HighUsageDetectionConfig) HighUsageThresholds() map[string][]HighUsageThreshold {
+	return cfg.Thresholds.effectiveThresholds()
+}
+
+type HighUsageThresholdFields struct {
+	ThresholdAddEventWindow          time.Duration `mapstructure:"threshold_add_event_window"`
+	ThresholdAddEventCount           uint32        `mapstructure:"threshold_add_event_count"`
+	ThresholdAddMediaEventWindow     time.Duration `mapstructure:"threshold_add_media_event_window"`
+	ThresholdAddMediaEventCount      uint32        `mapstructure:"threshold_add_media_event_count"`
+	ThresholdCreateMediaStreamWindow time.Duration `mapstructure:"threshold_create_media_stream_window"`
+	ThresholdCreateMediaStreamCount  uint32        `mapstructure:"threshold_create_media_stream_count"`
+}
+
+func (fields HighUsageThresholdFields) effectiveThresholds() map[string][]HighUsageThreshold {
+	result := make(map[string][]HighUsageThreshold)
+
+	addThreshold := func(name string, window time.Duration, count uint32) {
+		if window <= 0 || count == 0 || name == "" {
+			return
+		}
+		result[name] = append(result[name], HighUsageThreshold{
+			Window: window,
+			Count:  count,
+		})
+	}
+
+	addThreshold("event", fields.ThresholdAddEventWindow, fields.ThresholdAddEventCount)
+	addThreshold("media_event", fields.ThresholdAddMediaEventWindow, fields.ThresholdAddMediaEventCount)
+	addThreshold("create_media_stream", fields.ThresholdCreateMediaStreamWindow, fields.ThresholdCreateMediaStreamCount)
+
+	if len(result) == 0 {
+		return nil
+	}
+	return result
+}
+
+type HighUsageThreshold struct {
+	Window time.Duration
+	Count  uint32
+}
+
 type DebugEndpointsConfig struct {
 	Cache           bool
 	Memory          bool

core/config/config_test.go

@@ -3,6 +3,7 @@ package config_test
 import (
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -134,3 +135,64 @@ func TestConfig_ChainProvidersDoNotLog(t *testing.T) {
 	)
 	require.NotContains(logOutput, "CHAINS", "Expected CHAINS to be omitted from logOutput `%v`", logOutput)
 }
+
+func TestHighUsageDetectionConfig_DefaultThresholds(t *testing.T) {
+	cfg := config.HighUsageDetectionConfig{
+		Thresholds: config.HighUsageThresholdFields{
+			ThresholdAddEventWindow: time.Minute,
+			ThresholdAddEventCount:  100,
+		},
+	}
+
+	values := cfg.HighUsageThresholds()
+	require.Len(t, values, 1)
+	require.Len(t, values["event"], 1)
+	require.Equal(t, time.Minute, values["event"][0].Window)
+	require.Equal(t, uint32(100), values["event"][0].Count)
+}
+
+func TestHighUsageDetectionConfig_MultipleCallTypes(t *testing.T) {
+	cfg := config.HighUsageDetectionConfig{
+		Thresholds: config.HighUsageThresholdFields{
+			ThresholdAddEventWindow:          2 * time.Minute,
+			ThresholdAddEventCount:           200,
+			ThresholdAddMediaEventWindow:     45 * time.Second,
+			ThresholdAddMediaEventCount:      15,
+			ThresholdCreateMediaStreamWindow: 90 * time.Second,
+			ThresholdCreateMediaStreamCount:  25,
+		},
+	}
+
+	values := cfg.HighUsageThresholds()
+
+	require.Len(t, values["event"], 1)
+	require.Equal(t, 2*time.Minute, values["event"][0].Window)
+	require.Equal(t, uint32(200), values["event"][0].Count)
+	require.Len(t, values["media_event"], 1)
+	require.Equal(t, 45*time.Second, values["media_event"][0].Window)
+	require.Equal(t, uint32(15), values["media_event"][0].Count)
+	require.Len(t, values["create_media_stream"], 1)
+	require.Equal(t, 90*time.Second, values["create_media_stream"][0].Window)
+	require.Equal(t, uint32(25), values["create_media_stream"][0].Count)
+}
+
+func TestHighUsageDetectionConfig_InvalidEntriesIgnored(t *testing.T) {
+	cfg := config.HighUsageDetectionConfig{
+		Thresholds: config.HighUsageThresholdFields{
+			ThresholdAddEventWindow:          0,
+			ThresholdAddEventCount:           100,
+			ThresholdAddMediaEventWindow:     time.Minute,
+			ThresholdAddMediaEventCount:      0,
+			ThresholdCreateMediaStreamWindow: 30 * time.Second,
+			ThresholdCreateMediaStreamCount:  5,
+		},
+	}
+
+	values := cfg.HighUsageThresholds()
+
+	require.Equal(t, map[string][]config.HighUsageThreshold{
+		"create_media_stream": {
+			{Window: 30 * time.Second, Count: 5},
+		},
+	}, values)
+}

core/node/rpc/add_event.go

@@ -13,6 +13,7 @@ import (
 	"github.com/towns-protocol/towns/core/node/logging"
 	. "github.com/towns-protocol/towns/core/node/protocol"
 	rpcHeaders "github.com/towns-protocol/towns/core/node/rpc/headers"
+	"github.com/towns-protocol/towns/core/node/rpc/highusage"
 	"github.com/towns-protocol/towns/core/node/rules"
 	. "github.com/towns-protocol/towns/core/node/shared"
 )
@@ -50,11 +51,13 @@ func (s *Service) localAddEvent(
 
 	if err != nil {
 		return nil, err
-	} else {
-		return connect.NewResponse(&AddEventResponse{
-			NewEvents: newEvents,
-		}), nil
 	}
+
+	s.callRateMonitor.RecordCall(parsedEvent.Event.CreatorAddress, time.Now(), highusage.CallTypeEvent)
+
+	return connect.NewResponse(&AddEventResponse{
+		NewEvents: newEvents,
+	}), nil
 }
 
 // ensureStreamIsUpToDate returns the StreamView for the given StreamId that is up to date enough to

core/node/rpc/add_media_event.go

@@ -3,6 +3,7 @@ package rpc
 import (
 	"bytes"
 	"context"
+	"time"
 
 	"connectrpc.com/connect"
 	"github.com/ethereum/go-ethereum/common"
@@ -12,6 +13,7 @@ import (
 	. "github.com/towns-protocol/towns/core/node/events"
 	"github.com/towns-protocol/towns/core/node/logging"
 	. "github.com/towns-protocol/towns/core/node/protocol"
+	"github.com/towns-protocol/towns/core/node/rpc/highusage"
 	. "github.com/towns-protocol/towns/core/node/shared"
 )
 
@@ -73,6 +75,8 @@ func (s *Service) localAddMediaEvent(
 		return nil, AsRiverError(err).Func("localAddMediaEvent")
 	}
 
+	s.callRateMonitor.RecordCall(parsedEvent.Event.CreatorAddress, time.Now(), highusage.CallTypeMediaEvent)
+
 	return connect.NewResponse(&AddMediaEventResponse{
 		CreationCookie: &CreationCookie{
 			StreamId:          streamId[:],

core/node/rpc/create_media_stream.go

@@ -13,6 +13,7 @@ import (
 	"github.com/towns-protocol/towns/core/node/logging"
 	. "github.com/towns-protocol/towns/core/node/nodes"
 	. "github.com/towns-protocol/towns/core/node/protocol"
+	"github.com/towns-protocol/towns/core/node/rpc/highusage"
 	"github.com/towns-protocol/towns/core/node/rules"
 	. "github.com/towns-protocol/towns/core/node/shared"
 	"github.com/towns-protocol/towns/core/node/storage"
@@ -148,6 +149,8 @@ func (s *Service) createMediaStream(ctx context.Context, req *CreateMediaStreamR
 		return nil, AsRiverError(err).Func("createMediaStream")
 	}
 
+	s.callRateMonitor.RecordCall(parsedEvents[0].Event.CreatorAddress, time.Now(), highusage.CallTypeCreateMediaStream)
+
 	// add derived events
 	for _, de := range csRules.DerivedEvents {
 		_, err = s.AddEventPayload(ctx, de.StreamId, de.Payload, de.Tags)