feat(monitor/collectors/iss): 10s position poll + daily TLE refresh + propagator-ready collector

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/client.go

@@ -0,0 +1,121 @@
+// ©AngelaMos | 2026
+// client.go
+
+package iss
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"golang.org/x/time/rate"
+
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/httpx"
+)
+
+const (
+	defaultISSPositionURL = "https://api.wheretheiss.at"
+	defaultISSTLEURL      = "https://celestrak.org"
+	pathPosition          = "/v1/satellites/25544"
+	pathTLE               = "/NORAD/elements/gp.php"
+	defaultISSRate        = 200 * time.Millisecond
+	defaultISSBurst       = 5
+	defaultISSBudget      = 5
+	defaultISSBreaker     = 60 * time.Second
+)
+
+type Position struct {
+	Latitude  float64   `json:"latitude"`
+	Longitude float64   `json:"longitude"`
+	Altitude  float64   `json:"altitude"`
+	Velocity  float64   `json:"velocity"`
+	Timestamp int64     `json:"timestamp"`
+	FetchedAt time.Time `json:"fetched_at"`
+}
+
+type TLE struct {
+	Line1     string    `json:"line1"`
+	Line2     string    `json:"line2"`
+	FetchedAt time.Time `json:"fetched_at"`
+}
+
+type ClientConfig struct {
+	PositionBaseURL string
+	TLEBaseURL      string
+}
+
+type Client struct {
+	pos *httpx.Client
+	tle *httpx.Client
+}
+
+func NewClient(cfg ClientConfig) *Client {
+	if cfg.PositionBaseURL == "" {
+		cfg.PositionBaseURL = defaultISSPositionURL
+	}
+	if cfg.TLEBaseURL == "" {
+		cfg.TLEBaseURL = defaultISSTLEURL
+	}
+	return &Client{
+		pos: httpx.New(httpx.Config{
+			Name:                     "iss-position",
+			BaseURL:                  cfg.PositionBaseURL,
+			Rate:                     rate.Every(defaultISSRate),
+			Burst:                    defaultISSBurst,
+			ConsecutiveFailureBudget: defaultISSBudget,
+			BreakerTimeout:           defaultISSBreaker,
+		}),
+		tle: httpx.New(httpx.Config{
+			Name:                     "iss-tle",
+			BaseURL:                  cfg.TLEBaseURL,
+			Rate:                     rate.Every(time.Second),
+			Burst:                    1,
+			ConsecutiveFailureBudget: defaultISSBudget,
+			BreakerTimeout:           defaultISSBreaker,
+		}),
+	}
+}
+
+func (c *Client) FetchPosition(ctx context.Context) (Position, error) {
+	var p Position
+	if err := c.pos.GetJSON(ctx, pathPosition, nil, &p); err != nil {
+		return Position{}, fmt.Errorf("fetch iss position: %w", err)
+	}
+	p.FetchedAt = time.Now().UTC()
+	return p, nil
+}
+
+func (c *Client) FetchTLE(ctx context.Context) (TLE, error) {
+	q := map[string][]string{
+		"CATNR":  {"25544"},
+		"FORMAT": {"TLE"},
+	}
+	resp, err := c.tle.Get(ctx, pathTLE, q)
+	if err != nil {
+		return TLE{}, fmt.Errorf("fetch iss tle: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	body := strings.Builder{}
+	buf := make([]byte, 1024)
+	for {
+		n, rerr := resp.Body.Read(buf)
+		if n > 0 {
+			body.Write(buf[:n])
+		}
+		if rerr != nil {
+			break
+		}
+	}
+
+	lines := strings.Split(strings.TrimSpace(body.String()), "\n")
+	if len(lines) < 3 {
+		return TLE{}, fmt.Errorf("iss tle: expected 3 lines, got %d", len(lines))
+	}
+	return TLE{
+		Line1:     strings.TrimRight(lines[1], " \r"),
+		Line2:     strings.TrimRight(lines[2], " \r"),
+		FetchedAt: time.Now().UTC(),
+	}, nil
+}

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/collector.go

@@ -0,0 +1,119 @@
+// ©AngelaMos | 2026
+// collector.go
+
+package iss
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"time"
+
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/events"
+)
+
+const (
+	Name                    = "iss"
+	defaultPositionInterval = 10 * time.Second
+	defaultTLEInterval      = 24 * time.Hour
+)
+
+type Fetcher interface {
+	FetchPosition(ctx context.Context) (Position, error)
+	FetchTLE(ctx context.Context) (TLE, error)
+}
+
+type TLEStorer interface {
+	Save(ctx context.Context, tle TLE) error
+	Load(ctx context.Context) (TLE, bool, 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 {
+	PositionInterval time.Duration
+	TLEInterval      time.Duration
+	Fetcher          Fetcher
+	TLEStore         TLEStorer
+	Emitter          Emitter
+	State            StateRecorder
+	Logger           *slog.Logger
+}
+
+type Collector struct {
+	cfg    CollectorConfig
+	logger *slog.Logger
+}
+
+func NewCollector(cfg CollectorConfig) *Collector {
+	if cfg.PositionInterval <= 0 {
+		cfg.PositionInterval = defaultPositionInterval
+	}
+	if cfg.TLEInterval <= 0 {
+		cfg.TLEInterval = defaultTLEInterval
+	}
+	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 {
+	pos := time.NewTicker(c.cfg.PositionInterval)
+	defer pos.Stop()
+	tle := time.NewTicker(c.cfg.TLEInterval)
+	defer tle.Stop()
+
+	c.tickPosition(ctx)
+	c.tickTLE(ctx)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-pos.C:
+			c.tickPosition(ctx)
+		case <-tle.C:
+			c.tickTLE(ctx)
+		}
+	}
+}
+
+func (c *Collector) tickPosition(ctx context.Context) {
+	p, err := c.cfg.Fetcher.FetchPosition(ctx)
+	if err != nil {
+		c.logger.Warn("iss position fetch", "err", err)
+		_ = c.cfg.State.RecordError(ctx, Name, err.Error())
+		return
+	}
+	body, _ := json.Marshal(p)
+	c.cfg.Emitter.Emit(events.Event{
+		Topic:     events.TopicISSPosition,
+		Timestamp: p.FetchedAt,
+		Source:    Name,
+		Payload:   json.RawMessage(body),
+	})
+	_ = c.cfg.State.RecordSuccess(ctx, Name, 1)
+}
+
+func (c *Collector) tickTLE(ctx context.Context) {
+	tle, err := c.cfg.Fetcher.FetchTLE(ctx)
+	if err != nil {
+		c.logger.Warn("iss tle fetch", "err", err)
+		_ = c.cfg.State.RecordError(ctx, Name, err.Error())
+		return
+	}
+	if err := c.cfg.TLEStore.Save(ctx, tle); err != nil {
+		c.logger.Warn("iss tle save", "err", err)
+		_ = c.cfg.State.RecordError(ctx, Name, err.Error())
+	}
+}

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/collector_test.go

@@ -0,0 +1,183 @@
+// ©AngelaMos | 2026
+// collector_test.go
+
+package iss_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/iss"
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/events"
+)
+
+type fakeFetcher struct {
+	mu     sync.Mutex
+	pos    iss.Position
+	tle    iss.TLE
+	posErr error
+	tleErr error
+	posN   int
+	tleN   int
+}
+
+func (f *fakeFetcher) FetchPosition(_ context.Context) (iss.Position, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.posN++
+	return f.pos, f.posErr
+}
+
+func (f *fakeFetcher) FetchTLE(_ context.Context) (iss.TLE, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.tleN++
+	return f.tle, f.tleErr
+}
+
+type fakeStore struct {
+	mu    sync.Mutex
+	saves int
+	saved iss.TLE
+}
+
+func (s *fakeStore) Save(_ context.Context, tle iss.TLE) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.saves++
+	s.saved = tle
+	return nil
+}
+
+func (s *fakeStore) Load(_ context.Context) (iss.TLE, bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.saves == 0 {
+		return iss.TLE{}, false, nil
+	}
+	return s.saved, true, nil
+}
+
+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) Count() int {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	return len(e.events)
+}
+
+type recordingState struct {
+	mu        sync.Mutex
+	successes int
+	failures  int
+}
+
+func (s *recordingState) RecordSuccess(_ context.Context, _ string, _ int64) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.successes++
+	return nil
+}
+
+func (s *recordingState) RecordError(_ context.Context, _, _ string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.failures++
+	return nil
+}
+
+func TestCollector_PositionLoopEmitsAndRecords(t *testing.T) {
+	ftch := &fakeFetcher{
+		pos: iss.Position{Latitude: 10, Longitude: 20, Altitude: 420, Velocity: 27500, Timestamp: 1234, FetchedAt: time.Now().UTC()},
+		tle: iss.TLE{Line1: "1 25544U ...", Line2: "2 25544 ..."},
+	}
+	store := &fakeStore{}
+	emt := &fakeEmitter{}
+	st := &recordingState{}
+
+	c := iss.NewCollector(iss.CollectorConfig{
+		PositionInterval: 20 * time.Millisecond,
+		TLEInterval:      time.Hour,
+		Fetcher:          ftch,
+		TLEStore:         store,
+		Emitter:          emt,
+		State:            st,
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 70*time.Millisecond)
+	defer cancel()
+	_ = c.Run(ctx)
+
+	require.GreaterOrEqual(t, ftch.posN, 2, "position should fire multiple times")
+	require.GreaterOrEqual(t, emt.Count(), 2)
+	for _, ev := range emt.events {
+		require.Equal(t, events.TopicISSPosition, ev.Topic)
+	}
+	require.Greater(t, st.successes, 0)
+}
+
+func TestCollector_TLELoopSavesToStore(t *testing.T) {
+	ftch := &fakeFetcher{
+		pos: iss.Position{Latitude: 10, Longitude: 20, Altitude: 420, FetchedAt: time.Now().UTC()},
+		tle: iss.TLE{Line1: "1 25544U test", Line2: "2 25544 test"},
+	}
+	store := &fakeStore{}
+	emt := &fakeEmitter{}
+	st := &recordingState{}
+
+	c := iss.NewCollector(iss.CollectorConfig{
+		PositionInterval: 50 * time.Millisecond,
+		TLEInterval:      30 * time.Millisecond,
+		Fetcher:          ftch,
+		TLEStore:         store,
+		Emitter:          emt,
+		State:            st,
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	_ = c.Run(ctx)
+
+	require.GreaterOrEqual(t, store.saves, 2, "TLE should refresh multiple times in 100ms with 30ms interval")
+	require.Equal(t, "1 25544U test", store.saved.Line1)
+}
+
+func TestCollector_PositionFetchErrorRecordsState(t *testing.T) {
+	ftch := &fakeFetcher{
+		posErr: errors.New("upstream 503"),
+		tle:    iss.TLE{Line1: "1 ...", Line2: "2 ..."},
+	}
+	store := &fakeStore{}
+	emt := &fakeEmitter{}
+	st := &recordingState{}
+
+	c := iss.NewCollector(iss.CollectorConfig{
+		PositionInterval: 20 * time.Millisecond,
+		TLEInterval:      time.Hour,
+		Fetcher:          ftch,
+		TLEStore:         store,
+		Emitter:          emt,
+		State:            st,
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	_ = c.Run(ctx)
+
+	require.Equal(t, 0, emt.Count())
+	require.Greater(t, st.failures, 0)
+}

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator.go

@@ -27,7 +27,7 @@ func LoadTLE(line1, line2 string) (Sat, error) {
 	return Sat{inner: s}, nil
 }
 
-func Position(s Sat, t time.Time) (lat, lon, altKm float64) {
+func Propagate(s Sat, t time.Time) (lat, lon, altKm float64) {
 	t = t.UTC()
 	pos, _ := satellite.Propagate(s.inner, t.Year(), int(t.Month()), t.Day(), t.Hour(), t.Minute(), t.Second())
 	gmst := satellite.GSTimeFromDate(t.Year(), int(t.Month()), t.Day(), t.Hour(), t.Minute(), t.Second())

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator_test.go

@@ -29,7 +29,7 @@ func TestPropagator_PositionWithinReasonableBounds(t *testing.T) {
 	require.NoError(t, err)
 
 	when := time.Date(2026, 5, 2, 3, 0, 0, 0, time.UTC)
-	lat, lon, alt := iss.Position(sat, when)
+	lat, lon, alt := iss.Propagate(sat, when)
 
 	require.True(t, lat >= -90 && lat <= 90, "lat=%f", lat)
 	require.True(t, lon >= -180 && lon <= 180, "lon=%f", lon)

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/tle_store.go

@@ -0,0 +1,53 @@
+// ©AngelaMos | 2026
+// tle_store.go
+
+package iss
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	tleKey = "iss:tle"
+	tleTTL = 24 * time.Hour
+)
+
+type TLEStore struct {
+	rdb *redis.Client
+}
+
+func NewTLEStore(rdb *redis.Client) *TLEStore {
+	return &TLEStore{rdb: rdb}
+}
+
+func (s *TLEStore) Save(ctx context.Context, tle TLE) error {
+	body, err := json.Marshal(tle)
+	if err != nil {
+		return fmt.Errorf("marshal tle: %w", err)
+	}
+	if err := s.rdb.Set(ctx, tleKey, body, tleTTL).Err(); err != nil {
+		return fmt.Errorf("save iss tle: %w", err)
+	}
+	return nil
+}
+
+func (s *TLEStore) Load(ctx context.Context) (TLE, bool, error) {
+	body, err := s.rdb.Get(ctx, tleKey).Bytes()
+	if errors.Is(err, redis.Nil) {
+		return TLE{}, false, nil
+	}
+	if err != nil {
+		return TLE{}, false, fmt.Errorf("load iss tle: %w", err)
+	}
+	var tle TLE
+	if err := json.Unmarshal(body, &tle); err != nil {
+		return TLE{}, false, fmt.Errorf("unmarshal tle: %w", err)
+	}
+	return tle, true, nil
+}