From bc9c2d06ebd6ab5e52585d397289709909a8a267 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Fri, 1 May 2026 07:03:47 -0400 Subject: [PATCH] feat(monitor/backend): events, ratelimit, bus core packages with TDD --- .../backend/go.mod | 5 + .../backend/go.sum | 2 + .../backend/internal/bus/bus.go | 85 ++++++++++++++++ .../backend/internal/bus/bus_test.go | 99 +++++++++++++++++++ .../backend/internal/bus/persister.go | 23 +++++ .../backend/internal/events/event.go | 13 +++ .../backend/internal/events/event_test.go | 48 +++++++++ .../backend/internal/events/topic.go | 60 +++++++++++ .../backend/internal/ratelimit/limiter.go | 70 +++++++++++++ .../internal/ratelimit/limiter_test.go | 65 ++++++++++++ 10 files changed, 470 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus_test.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/persister.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event_test.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/topic.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod index 1f8967f3..72ee3fd4 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod @@ -15,6 +15,8 @@ require ( github.com/knadh/koanf/v2 v2.1.2 github.com/lestrrat-go/jwx/v3 v3.0.12 github.com/redis/go-redis/v9 v9.7.3 + github.com/sony/gobreaker/v2 v2.4.0 + github.com/stretchr/testify v1.11.1 go.opentelemetry.io/otel v1.41.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.33.0 go.opentelemetry.io/otel/sdk v1.35.0 @@ -27,6 +29,7 @@ require ( require ( github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect @@ -52,6 +55,7 @@ require ( github.com/lestrrat-go/option/v2 v2.0.0 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/segmentio/asm v1.2.1 // indirect github.com/valyala/fastjson v1.6.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect @@ -66,4 +70,5 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20241209162323-e6fa225c2576 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576 // indirect google.golang.org/protobuf v1.36.8 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum index 2cc02b59..7d9a840f 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum @@ -108,6 +108,8 @@ github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0t github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= +github.com/sony/gobreaker/v2 v2.4.0 h1:g2KJRW1Ubty3+ZOcSEUN7K+REQJdN6yo6XvaML+jptg= +github.com/sony/gobreaker/v2 v2.4.0/go.mod h1:pTyFJgcZ3h2tdQVLZZruK2C0eoFL1fb/G83wK1ZQl+s= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus.go new file mode 100644 index 00000000..6ea8445c --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus.go @@ -0,0 +1,85 @@ +// ©AngelaMos | 2026 +// bus.go + +package bus + +import ( + "context" + "encoding/json" + "log/slog" + "sync/atomic" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +const defaultBufferSize = 512 + +type Config struct { + BufferSize int + Persister Persister + Broadcaster Broadcaster + Logger *slog.Logger +} + +type Bus struct { + ch chan events.Event + persister Persister + broadcaster Broadcaster + logger *slog.Logger + dropped atomic.Uint64 +} + +func New(cfg Config) *Bus { + if cfg.BufferSize <= 0 { + cfg.BufferSize = defaultBufferSize + } + logger := cfg.Logger + if logger == nil { + logger = slog.Default() + } + return &Bus{ + ch: make(chan events.Event, cfg.BufferSize), + persister: cfg.Persister, + broadcaster: cfg.Broadcaster, + logger: logger, + } +} + +func (b *Bus) Emit(ev events.Event) { + select { + case b.ch <- ev: + default: + b.dropped.Add(1) + b.logger.Warn("event bus full, dropped", + "topic", ev.Topic, "source", ev.Source) + } +} + +func (b *Bus) DroppedCount() uint64 { + return b.dropped.Load() +} + +func (b *Bus) Run(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + case ev := <-b.ch: + if b.persister != nil { + if err := b.persister.Save(ctx, ev); err != nil { + b.logger.Error("persist event failed", + "err", err, "topic", ev.Topic) + } + } + if b.broadcaster != nil { + payload, err := json.Marshal(ev.Payload) + if err != nil { + b.logger.Error("marshal payload failed", + "err", err, "topic", ev.Topic) + continue + } + b.broadcaster.Broadcast(string(ev.Topic), payload) + } + } + } +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus_test.go new file mode 100644 index 00000000..f2115a06 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/bus_test.go @@ -0,0 +1,99 @@ +// ©AngelaMos | 2026 +// bus_test.go + +package bus_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/bus" + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +type capturePersister struct { + mu sync.Mutex + saved []events.Event +} + +func (c *capturePersister) Save(_ context.Context, ev events.Event) error { + c.mu.Lock() + defer c.mu.Unlock() + c.saved = append(c.saved, ev) + return nil +} + +func (c *capturePersister) Count() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.saved) +} + +type captureBroadcaster struct { + mu sync.Mutex + sent []bus.Broadcast +} + +func (b *captureBroadcaster) Broadcast(topic string, payload []byte) { + b.mu.Lock() + defer b.mu.Unlock() + b.sent = append(b.sent, bus.Broadcast{Topic: topic, Payload: payload}) +} + +func (b *captureBroadcaster) Count() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.sent) +} + +func TestBus_EmitFlowsThroughDispatcher(t *testing.T) { + persist := &capturePersister{} + cast := &captureBroadcaster{} + + b := bus.New(bus.Config{ + BufferSize: 16, + Persister: persist, + Broadcaster: cast, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { done <- b.Run(ctx) }() + + now := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) + b.Emit(events.Event{ + Topic: events.TopicHeartbeat, + Timestamp: now, + Source: "heartbeat", + Payload: map[string]string{"ts": now.Format(time.RFC3339Nano)}, + }) + + require.Eventually(t, func() bool { + return persist.Count() == 1 && cast.Count() == 1 + }, 2*time.Second, 10*time.Millisecond) + + cancel() + require.ErrorIs(t, <-done, context.Canceled) +} + +func TestBus_DropOnFullBuffer(t *testing.T) { + persist := &capturePersister{} + cast := &captureBroadcaster{} + + b := bus.New(bus.Config{ + BufferSize: 1, + Persister: persist, + Broadcaster: cast, + }) + + for i := 0; i < 50; i++ { + b.Emit(events.Event{Topic: events.TopicHeartbeat, Source: "stress"}) + } + require.Greater(t, b.DroppedCount(), uint64(0)) +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/persister.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/persister.go new file mode 100644 index 00000000..1472faa5 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/bus/persister.go @@ -0,0 +1,23 @@ +// ©AngelaMos | 2026 +// persister.go + +package bus + +import ( + "context" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +type Persister interface { + Save(ctx context.Context, ev events.Event) error +} + +type Broadcaster interface { + Broadcast(topic string, payload []byte) +} + +type Broadcast struct { + Topic string + Payload []byte +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event.go new file mode 100644 index 00000000..eb07a78b --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event.go @@ -0,0 +1,13 @@ +// ©AngelaMos | 2026 +// event.go + +package events + +import "time" + +type Event struct { + Topic Topic `json:"topic"` + Timestamp time.Time `json:"timestamp"` + Source string `json:"source"` + Payload any `json:"payload"` +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event_test.go new file mode 100644 index 00000000..88072321 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/event_test.go @@ -0,0 +1,48 @@ +// ©AngelaMos | 2026 +// event_test.go + +package events_test + +import ( + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +func TestEvent_JSONRoundTrip(t *testing.T) { + now := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) + original := events.Event{ + Topic: events.TopicHeartbeat, + Timestamp: now, + Source: "heartbeat", + Payload: map[string]any{"ts": now.Format(time.RFC3339Nano)}, + } + + raw, err := json.Marshal(original) + require.NoError(t, err) + + var decoded events.Event + require.NoError(t, json.Unmarshal(raw, &decoded)) + + require.Equal(t, original.Topic, decoded.Topic) + require.Equal(t, original.Source, decoded.Source) + require.True(t, original.Timestamp.Equal(decoded.Timestamp)) +} + +func TestTopic_IsValid(t *testing.T) { + require.True(t, events.TopicHeartbeat.IsValid()) + require.True(t, events.TopicCVENew.IsValid()) + require.False(t, events.Topic("bogus").IsValid()) +} + +func TestAllTopics_AllValid(t *testing.T) { + all := events.AllTopics() + require.NotEmpty(t, all) + for _, top := range all { + require.True(t, top.IsValid(), "topic %q should be valid", top) + } +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/topic.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/topic.go new file mode 100644 index 00000000..27e2c603 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/events/topic.go @@ -0,0 +1,60 @@ +// ©AngelaMos | 2026 +// topic.go + +package events + +type Topic string + +const ( + TopicHeartbeat Topic = "heartbeat" + TopicScanFirehose Topic = "scan_firehose" + TopicInternetOutage Topic = "internet_outage" + TopicBGPHijack Topic = "bgp_hijack" + TopicCVENew Topic = "cve_new" + TopicCVEUpdated Topic = "cve_updated" + TopicEPSS Topic = "epss" + TopicKEVAdded Topic = "kev_added" + TopicRansomwareVictim Topic = "ransomware_victim" + TopicCoinbasePrice Topic = "coinbase_price" + TopicEarthquake Topic = "earthquake" + TopicSpaceWeather Topic = "space_weather" + TopicWikipediaITN Topic = "wiki_itn" + TopicGDELTSpike Topic = "gdelt_spike" + TopicISSPosition Topic = "iss_position" + TopicCollectorState Topic = "collector_state" +) + +func (t Topic) String() string { return string(t) } + +func (t Topic) IsValid() bool { + switch t { + case TopicHeartbeat, TopicScanFirehose, TopicInternetOutage, TopicBGPHijack, + TopicCVENew, TopicCVEUpdated, TopicEPSS, TopicKEVAdded, + TopicRansomwareVictim, TopicCoinbasePrice, TopicEarthquake, + TopicSpaceWeather, TopicWikipediaITN, TopicGDELTSpike, + TopicISSPosition, TopicCollectorState: + return true + } + return false +} + +func AllTopics() []Topic { + return []Topic{ + TopicHeartbeat, + TopicScanFirehose, + TopicInternetOutage, + TopicBGPHijack, + TopicCVENew, + TopicCVEUpdated, + TopicEPSS, + TopicKEVAdded, + TopicRansomwareVictim, + TopicCoinbasePrice, + TopicEarthquake, + TopicSpaceWeather, + TopicWikipediaITN, + TopicGDELTSpike, + TopicISSPosition, + TopicCollectorState, + } +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter.go new file mode 100644 index 00000000..5e842475 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter.go @@ -0,0 +1,70 @@ +// ©AngelaMos | 2026 +// limiter.go + +package ratelimit + +import ( + "context" + "log/slog" + "time" + + "github.com/sony/gobreaker/v2" + "golang.org/x/time/rate" +) + +type Config struct { + Name string + Rate rate.Limit + Burst int + ConsecutiveFailureBudget uint32 + BreakerTimeout time.Duration + OnStateChange func(name string, from, to gobreaker.State) +} + +type Limiter[T any] struct { + name string + rl *rate.Limiter + cb *gobreaker.CircuitBreaker[T] +} + +func New[T any](cfg Config) *Limiter[T] { + if cfg.ConsecutiveFailureBudget == 0 { + cfg.ConsecutiveFailureBudget = 5 + } + if cfg.BreakerTimeout == 0 { + cfg.BreakerTimeout = 30 * time.Second + } + settings := gobreaker.Settings{ + Name: cfg.Name, + Timeout: cfg.BreakerTimeout, + ReadyToTrip: func(c gobreaker.Counts) bool { + return c.ConsecutiveFailures >= cfg.ConsecutiveFailureBudget + }, + OnStateChange: func(name string, from, to gobreaker.State) { + slog.Warn("circuit breaker state change", + "name", name, "from", from.String(), "to", to.String()) + if cfg.OnStateChange != nil { + cfg.OnStateChange(name, from, to) + } + }, + } + return &Limiter[T]{ + name: cfg.Name, + rl: rate.NewLimiter(cfg.Rate, cfg.Burst), + cb: gobreaker.NewCircuitBreaker[T](settings), + } +} + +func (l *Limiter[T]) Do(ctx context.Context, fn func(context.Context) (T, error)) (T, error) { + var zero T + if err := l.rl.Wait(ctx); err != nil { + return zero, err + } + return l.cb.Execute(func() (T, error) { return fn(ctx) }) +} + +func (l *Limiter[T]) AdjustRate(newRate rate.Limit) { + l.rl.SetLimit(newRate) +} + +func (l *Limiter[T]) Name() string { return l.name } diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter_test.go new file mode 100644 index 00000000..8299b139 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/ratelimit/limiter_test.go @@ -0,0 +1,65 @@ +// ©AngelaMos | 2026 +// limiter_test.go + +package ratelimit_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/sony/gobreaker/v2" + "github.com/stretchr/testify/require" + "golang.org/x/time/rate" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/ratelimit" +) + +func TestLimiter_AllowsAtNominalRate(t *testing.T) { + lim := ratelimit.New[string](ratelimit.Config{ + Name: "test-allow", + Rate: rate.Every(20 * time.Millisecond), + Burst: 4, + ConsecutiveFailureBudget: 5, + BreakerTimeout: 100 * time.Millisecond, + }) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + calls := 0 + for i := 0; i < 4; i++ { + _, err := lim.Do(ctx, func(_ context.Context) (string, error) { + calls++ + return "ok", nil + }) + require.NoError(t, err) + } + require.Equal(t, 4, calls) +} + +func TestLimiter_OpensBreakerOnRepeatedFailures(t *testing.T) { + lim := ratelimit.New[string](ratelimit.Config{ + Name: "test-fail", + Rate: rate.Every(2 * time.Millisecond), + Burst: 10, + ConsecutiveFailureBudget: 3, + BreakerTimeout: 50 * time.Millisecond, + }) + + ctx := context.Background() + failure := errors.New("upstream broken") + + for i := 0; i < 3; i++ { + _, _ = lim.Do(ctx, func(_ context.Context) (string, error) { return "", failure }) + } + + called := false + _, err := lim.Do(ctx, func(_ context.Context) (string, error) { + called = true + return "", nil + }) + require.False(t, called, "should not have called function — breaker should be open") + require.ErrorIs(t, err, gobreaker.ErrOpenState) +}