diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/dshield/collector.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/dshield/collector.go new file mode 100644 index 00000000..ff2c4827 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/dshield/collector.go @@ -0,0 +1,117 @@ +// ©AngelaMos | 2026 +// collector.go + +package dshield + +import ( + "context" + "encoding/json" + "log/slog" + "time" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +const ( + Name = "dshield" + defaultInterval = time.Hour +) + +type Fetcher interface { + FetchAll(ctx context.Context) ([]SnapshotPayload, error) +} + +type Persister interface { + PutSnapshot(ctx context.Context, ts time.Time, kind string, payload json.RawMessage) error +} + +type Emitter interface { + Emit(ev events.Event) +} + +type StateRecorder interface { + RecordSuccess(ctx context.Context, name string, eventCount int64) error + RecordError(ctx context.Context, name, errMsg string) error +} + +type CollectorConfig struct { + Interval time.Duration + Fetcher Fetcher + Persister Persister + Emitter Emitter + State StateRecorder + Logger *slog.Logger +} + +type Collector struct { + cfg CollectorConfig + logger *slog.Logger +} + +func NewCollector(cfg CollectorConfig) *Collector { + if cfg.Interval <= 0 { + cfg.Interval = defaultInterval + } + if cfg.Logger == nil { + cfg.Logger = slog.Default() + } + return &Collector{cfg: cfg, logger: cfg.Logger} +} + +func (c *Collector) Name() string { return Name } + +func (c *Collector) Run(ctx context.Context) error { + ticker := time.NewTicker(c.cfg.Interval) + defer ticker.Stop() + + c.tick(ctx) + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + c.tick(ctx) + } + } +} + +func (c *Collector) tick(ctx context.Context) { + snaps, err := c.cfg.Fetcher.FetchAll(ctx) + if err != nil { + c.logger.Warn("dshield fetch failed", "err", err) + _ = c.cfg.State.RecordError(ctx, Name, err.Error()) + return + } + + now := time.Now().UTC() + tsRaw, err := json.Marshal(now.Format(time.RFC3339Nano)) + if err != nil { + c.logger.Error("dshield marshal ts", "err", err) + _ = c.cfg.State.RecordError(ctx, Name, err.Error()) + return + } + + merged := map[string]json.RawMessage{"ts": tsRaw} + for _, s := range snaps { + if perr := c.cfg.Persister.PutSnapshot(ctx, now, s.Kind, s.Payload); perr != nil { + c.logger.Warn("dshield persist failed", "kind", s.Kind, "err", perr) + } + merged[s.Kind] = s.Payload + } + + body, err := json.Marshal(merged) + if err != nil { + c.logger.Error("dshield marshal merged", "err", err) + _ = c.cfg.State.RecordError(ctx, Name, err.Error()) + return + } + + c.cfg.Emitter.Emit(events.Event{ + Topic: events.TopicScanFirehose, + Timestamp: now, + Source: Name, + Payload: json.RawMessage(body), + }) + _ = c.cfg.State.RecordSuccess(ctx, Name, 1) +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/dshield/collector_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/dshield/collector_test.go new file mode 100644 index 00000000..4e43288a --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/dshield/collector_test.go @@ -0,0 +1,177 @@ +// ©AngelaMos | 2026 +// collector_test.go + +package dshield_test + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/dshield" + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +type fakeFetcher struct { + mu sync.Mutex + calls int + err error +} + +func (f *fakeFetcher) FetchAll(_ context.Context) ([]dshield.SnapshotPayload, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls++ + if f.err != nil { + return nil, f.err + } + return []dshield.SnapshotPayload{ + {Kind: dshield.KindTopPorts, Payload: json.RawMessage(`[{"port":22}]`)}, + {Kind: dshield.KindTopIPs, Payload: json.RawMessage(`[{"ip":"1.2.3.4"}]`)}, + {Kind: dshield.KindDailySummary, Payload: json.RawMessage(`[{"records":42}]`)}, + }, nil +} + +func (f *fakeFetcher) Calls() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls +} + +type persistedRow struct { + Kind string + Body json.RawMessage +} + +type fakePersister struct { + mu sync.Mutex + rows []persistedRow +} + +func (p *fakePersister) PutSnapshot(_ context.Context, _ time.Time, kind string, body json.RawMessage) error { + p.mu.Lock() + defer p.mu.Unlock() + p.rows = append(p.rows, persistedRow{Kind: kind, Body: body}) + return nil +} + +func (p *fakePersister) Rows() []persistedRow { + p.mu.Lock() + defer p.mu.Unlock() + out := make([]persistedRow, len(p.rows)) + copy(out, p.rows) + return out +} + +type fakeEmitter struct { + mu sync.Mutex + events []events.Event +} + +func (e *fakeEmitter) Emit(ev events.Event) { + e.mu.Lock() + defer e.mu.Unlock() + e.events = append(e.events, ev) +} + +func (e *fakeEmitter) Events() []events.Event { + e.mu.Lock() + defer e.mu.Unlock() + out := make([]events.Event, len(e.events)) + copy(out, e.events) + return out +} + +type recordingState struct { + mu sync.Mutex + successes int + successN int64 + failures int + lastErr string +} + +func (s *recordingState) RecordSuccess(_ context.Context, _ string, n int64) error { + s.mu.Lock() + defer s.mu.Unlock() + s.successes++ + s.successN += n + return nil +} + +func (s *recordingState) RecordError(_ context.Context, _, msg string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.failures++ + s.lastErr = msg + return nil +} + +func TestCollector_TickPersistsAllAndEmitsScanFirehose(t *testing.T) { + ftch := &fakeFetcher{} + prst := &fakePersister{} + emt := &fakeEmitter{} + st := &recordingState{} + + c := dshield.NewCollector(dshield.CollectorConfig{ + Interval: 20 * time.Millisecond, + Fetcher: ftch, + Persister: prst, + Emitter: emt, + State: st, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 70*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.GreaterOrEqual(t, ftch.Calls(), 2) + require.GreaterOrEqual(t, len(prst.Rows()), 6) + + evs := emt.Events() + require.GreaterOrEqual(t, len(evs), 2) + for _, ev := range evs { + require.Equal(t, events.TopicScanFirehose, ev.Topic) + require.Equal(t, dshield.Name, ev.Source) + body, ok := ev.Payload.(json.RawMessage) + require.True(t, ok, "payload must be json.RawMessage") + + var merged map[string]json.RawMessage + require.NoError(t, json.Unmarshal(body, &merged)) + require.Contains(t, merged, dshield.KindTopPorts) + require.Contains(t, merged, dshield.KindTopIPs) + require.Contains(t, merged, dshield.KindDailySummary) + require.Contains(t, merged, "ts") + } + + require.Greater(t, st.successes, 0) + require.Equal(t, 0, st.failures) +} + +func TestCollector_FetchErrorRecordsState(t *testing.T) { + ftch := &fakeFetcher{err: errors.New("upstream 503")} + prst := &fakePersister{} + emt := &fakeEmitter{} + st := &recordingState{} + + c := dshield.NewCollector(dshield.CollectorConfig{ + Interval: 20 * time.Millisecond, + Fetcher: ftch, + Persister: prst, + Emitter: emt, + State: st, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.Equal(t, 0, len(prst.Rows())) + require.Equal(t, 0, len(emt.Events())) + require.Greater(t, st.failures, 0) + require.Contains(t, st.lastErr, "upstream 503") +}