From 89f1c102492b013425f9e783e6a2fe6b7b68cad1 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sat, 2 May 2026 03:32:16 -0400 Subject: [PATCH] feat(monitor/collectors/coinbase): collector Run loop ties dial+readloop+aggregator+repo+throttled-emit --- .../internal/collectors/coinbase/collector.go | 169 +++++++++++++++++ .../collectors/coinbase/collector_test.go | 177 ++++++++++++++++++ 2 files changed, 346 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector.go new file mode 100644 index 00000000..457a51aa --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector.go @@ -0,0 +1,169 @@ +// ©AngelaMos | 2026 +// collector.go + +package coinbase + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "sync" + "time" + + "github.com/shopspring/decimal" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +const ( + Name = "coinbase" + defaultURL = "wss://advanced-trade-ws.coinbase.com" + defaultThrottle = 250 * time.Millisecond +) + +type Repository interface { + InsertTick(ctx context.Context, t Tick) error + UpsertMinute(ctx context.Context, b MinuteBar) error + LatestTick(ctx context.Context, symbol string) (Tick, error) + History1h(ctx context.Context, symbol string) ([]MinuteBar, 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 { + URL string + ProductIDs []string + Repo Repository + Emitter Emitter + State StateRecorder + Dialer Dialer + Throttle time.Duration + Reconnect ReconnectConfig + Logger *slog.Logger +} + +type Collector struct { + cfg CollectorConfig + dialer Dialer + logger *slog.Logger + mu sync.Mutex + lastEmit map[string]time.Time +} + +func NewCollector(cfg CollectorConfig) *Collector { + if cfg.URL == "" { + cfg.URL = defaultURL + } + if len(cfg.ProductIDs) == 0 { + cfg.ProductIDs = []string{defaultProductBTC, defaultProductETH} + } + if cfg.Throttle <= 0 { + cfg.Throttle = defaultThrottle + } + if cfg.Logger == nil { + cfg.Logger = slog.Default() + } + + d := cfg.Dialer + if d == nil { + d = NewWSDialer(DialerConfig{URL: cfg.URL, ProductIDs: cfg.ProductIDs}) + } + + return &Collector{ + cfg: cfg, + dialer: d, + logger: cfg.Logger, + lastEmit: make(map[string]time.Time), + } +} + +func (c *Collector) Name() string { return Name } + +func (c *Collector) Run(ctx context.Context) error { + err := Reconnect(ctx, c.dialer, c.cfg.Reconnect, c.handleConn) + if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + _ = c.cfg.State.RecordError(ctx, Name, err.Error()) + } + return err +} + +func (c *Collector) handleConn(ctx context.Context, conn *Conn) error { + seq := NewSequencer() + agg := NewAggregator() + count := int64(0) + + loopErr := ReadLoop(ctx, conn, seq, func(hctx context.Context, f Frame) error { + switch f.Kind { + case FrameTypeTicker, FrameTypeSnapshot: + for _, tk := range f.Tickers { + tick := Tick{ + Symbol: tk.ProductID, + TS: tk.Time.UTC(), + Price: tk.Price, + Volume24h: tk.Volume24h, + } + if err := c.cfg.Repo.InsertTick(hctx, tick); err != nil { + c.logger.Warn("insert tick", "symbol", tick.Symbol, "err", err) + continue + } + if closed, _ := agg.Push(tick); closed != nil { + if err := c.cfg.Repo.UpsertMinute(hctx, *closed); err != nil { + c.logger.Warn("upsert minute", "symbol", closed.Symbol, "minute", closed.Minute, "err", err) + } + } + if c.shouldEmit(tick.Symbol) { + c.emitTick(tick) + count++ + } + } + } + return nil + }) + + if loopErr == nil || errors.Is(loopErr, ErrSequenceGap) { + _ = c.cfg.State.RecordSuccess(ctx, Name, count) + } + return loopErr +} + +func (c *Collector) shouldEmit(symbol string) bool { + c.mu.Lock() + defer c.mu.Unlock() + last, ok := c.lastEmit[symbol] + now := time.Now() + if !ok || now.Sub(last) >= c.cfg.Throttle { + c.lastEmit[symbol] = now + return true + } + return false +} + +type tickPayload struct { + Symbol string `json:"symbol"` + TS time.Time `json:"ts"` + Price decimal.Decimal `json:"price"` + Volume24h decimal.Decimal `json:"volume_24h"` +} + +func (c *Collector) emitTick(t Tick) { + body, _ := json.Marshal(tickPayload{ + Symbol: t.Symbol, + TS: t.TS, + Price: t.Price, + Volume24h: t.Volume24h, + }) + c.cfg.Emitter.Emit(events.Event{ + Topic: events.TopicCoinbasePrice, + Timestamp: t.TS, + Source: Name, + Payload: json.RawMessage(body), + }) +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector_test.go new file mode 100644 index 00000000..f36258a6 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/collector_test.go @@ -0,0 +1,177 @@ +// ©AngelaMos | 2026 +// collector_test.go + +package coinbase_test + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/coinbase" + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +type fakeRepo struct { + mu sync.Mutex + ticks []coinbase.Tick + minutes []coinbase.MinuteBar +} + +func (r *fakeRepo) InsertTick(_ context.Context, t coinbase.Tick) error { + r.mu.Lock() + defer r.mu.Unlock() + r.ticks = append(r.ticks, t) + return nil +} + +func (r *fakeRepo) UpsertMinute(_ context.Context, b coinbase.MinuteBar) error { + r.mu.Lock() + defer r.mu.Unlock() + r.minutes = append(r.minutes, b) + return nil +} + +func (r *fakeRepo) LatestTick(_ context.Context, _ string) (coinbase.Tick, error) { + return coinbase.Tick{}, nil +} + +func (r *fakeRepo) History1h(_ context.Context, _ string) ([]coinbase.MinuteBar, error) { + return nil, nil +} + +func (r *fakeRepo) Tickers() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.ticks) +} + +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 fakeState struct { + mu sync.Mutex + successes int + failures int +} + +func (s *fakeState) RecordSuccess(_ context.Context, _ string, _ int64) error { + s.mu.Lock() + defer s.mu.Unlock() + s.successes++ + return nil +} + +func (s *fakeState) RecordError(_ context.Context, _, _ string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.failures++ + return nil +} + +func tickerFrame(seq int64, productID, price string, ts time.Time) []byte { + body, _ := json.Marshal(map[string]any{ + "channel": "ticker", + "sequence_num": seq, + "timestamp": ts.Format(time.RFC3339Nano), + "events": []any{ + map[string]any{ + "type": "update", + "tickers": []any{ + map[string]any{ + "type": "ticker", + "product_id": productID, + "price": price, + "volume_24_h": "1.0", + "time": ts.Format(time.RFC3339Nano), + }, + }, + }, + }, + }) + return body +} + +func TestCollector_RunPersistsAndEmits(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + frames := [][]byte{loadFixture(t, "subscriptions.json")} + for i := 0; i < 6; i++ { + ts := now.Add(time.Duration(i) * time.Second) + frames = append(frames, tickerFrame(int64(i+1), "BTC-USD", fmt.Sprintf("4200%d.00", i), ts)) + } + fs := newFakeServer(t, frames...) + + repo := &fakeRepo{} + emt := &fakeEmitter{} + st := &fakeState{} + + c := coinbase.NewCollector(coinbase.CollectorConfig{ + URL: fs.URL(), + ProductIDs: []string{"BTC-USD"}, + Repo: repo, + Emitter: emt, + State: st, + Throttle: 10 * time.Millisecond, + Reconnect: coinbase.ReconnectConfig{ + InitialInterval: 5 * time.Millisecond, + MaxInterval: 20 * time.Millisecond, + }, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.GreaterOrEqual(t, repo.Tickers(), 3, "should persist most ticks") + require.GreaterOrEqual(t, emt.Count(), 1, "should emit at least one event after throttle") +} + +func TestCollector_GapTriggersReconnect(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + frames := [][]byte{ + loadFixture(t, "subscriptions.json"), + tickerFrame(100, "BTC-USD", "42000.00", now), + tickerFrame(500, "BTC-USD", "42100.00", now), + } + fs := newFakeServer(t, frames...) + + repo := &fakeRepo{} + emt := &fakeEmitter{} + st := &fakeState{} + + c := coinbase.NewCollector(coinbase.CollectorConfig{ + URL: fs.URL(), + ProductIDs: []string{"BTC-USD"}, + Repo: repo, + Emitter: emt, + State: st, + Throttle: 10 * time.Millisecond, + Reconnect: coinbase.ReconnectConfig{ + InitialInterval: 5 * time.Millisecond, + MaxInterval: 20 * time.Millisecond, + }, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 600*time.Millisecond) + defer cancel() + _ = c.Run(ctx) +}