diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/client.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/client.go new file mode 100644 index 00000000..364ca26e --- /dev/null +++ b/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 +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/collector.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/collector.go new file mode 100644 index 00000000..36b27b4c --- /dev/null +++ b/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()) + } +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/collector_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/collector_test.go new file mode 100644 index 00000000..07141ca3 --- /dev/null +++ b/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) +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator.go index 73e951a4..c446ab6f 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator.go +++ b/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()) diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator_test.go index ccb18a6c..3317e467 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/propagator_test.go +++ b/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) diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/tle_store.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/iss/tle_store.go new file mode 100644 index 00000000..c548e90e --- /dev/null +++ b/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 +}