feat(monitor/backend): events, ratelimit, bus core packages with TDD

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
 )

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=

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)
+			}
+		}
+	}
+}

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))
+}

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
+}

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"`
+}

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)
+	}
+}

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,
+	}
+}

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 }

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)
+}