From 10862314f02cc62b3f1381d244febc73b7b1de93 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sat, 2 May 2026 03:26:58 -0400 Subject: [PATCH] feat(monitor/collectors/coinbase): WS dialer + frame decoder for ticker/heartbeats/snapshot --- .../internal/collectors/coinbase/client.go | 191 ++++++++++++++++++ .../collectors/coinbase/client_test.go | 190 +++++++++++++++++ .../coinbase/testdata/heartbeats.json | 9 + .../coinbase/testdata/snapshot.json | 15 ++ .../coinbase/testdata/subscriptions.json | 9 + .../collectors/coinbase/testdata/ticker.json | 20 ++ 6 files changed, 434 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client_test.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/heartbeats.json create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/snapshot.json create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/subscriptions.json create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/ticker.json diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client.go new file mode 100644 index 00000000..0489b1e7 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client.go @@ -0,0 +1,191 @@ +// ©AngelaMos | 2026 +// client.go + +package coinbase + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/coder/websocket" + "github.com/shopspring/decimal" +) + +const ( + defaultProductBTC = "BTC-USD" + defaultProductETH = "ETH-USD" + channelTicker = "ticker" + channelHeartbeats = "heartbeats" + channelSubscriptions = "subscriptions" + subscribeWriteTimeout = 5 * time.Second + defaultReadFrameSize = 1 << 20 +) + +type FrameType int + +const ( + FrameTypeUnknown FrameType = iota + FrameTypeSubscriptions + FrameTypeTicker + FrameTypeSnapshot + FrameTypeHeartbeats +) + +type TickerEntry struct { + ProductID string `json:"product_id"` + Price decimal.Decimal `json:"price"` + Volume24h decimal.Decimal `json:"volume_24_h"` + Time time.Time `json:"time"` +} + +type Frame struct { + Kind FrameType + SequenceNum int64 + Timestamp time.Time + Tickers []TickerEntry + HeartbeatTime time.Time +} + +type DialerConfig struct { + URL string + ProductIDs []string + WriteTimeout time.Duration + ReadFrameMax int64 +} + +type Dialer interface { + Dial(ctx context.Context) (*Conn, error) +} + +type WSDialer struct { + cfg DialerConfig +} + +func NewWSDialer(cfg DialerConfig) *WSDialer { + if len(cfg.ProductIDs) == 0 { + cfg.ProductIDs = []string{defaultProductBTC, defaultProductETH} + } + if cfg.WriteTimeout <= 0 { + cfg.WriteTimeout = subscribeWriteTimeout + } + if cfg.ReadFrameMax <= 0 { + cfg.ReadFrameMax = defaultReadFrameSize + } + return &WSDialer{cfg: cfg} +} + +func (d *WSDialer) Dial(ctx context.Context) (*Conn, error) { + c, _, err := websocket.Dial(ctx, d.cfg.URL, nil) + if err != nil { + return nil, fmt.Errorf("dial coinbase ws: %w", err) + } + c.SetReadLimit(d.cfg.ReadFrameMax) + + conn := &Conn{c: c} + subCtx, cancel := context.WithTimeout(ctx, d.cfg.WriteTimeout) + defer cancel() + if err := conn.subscribe(subCtx, channelTicker, d.cfg.ProductIDs); err != nil { + _ = c.Close(websocket.StatusInternalError, "subscribe ticker") + return nil, err + } + if err := conn.subscribe(subCtx, channelHeartbeats, nil); err != nil { + _ = c.Close(websocket.StatusInternalError, "subscribe heartbeats") + return nil, err + } + return conn, nil +} + +type Conn struct { + c *websocket.Conn +} + +func (c *Conn) Close() error { + if c == nil || c.c == nil { + return nil + } + return c.c.Close(websocket.StatusNormalClosure, "") +} + +type subscribeMsg struct { + Type string `json:"type"` + Channel string `json:"channel"` + ProductIDs []string `json:"product_ids,omitempty"` +} + +func (c *Conn) subscribe(ctx context.Context, channel string, productIDs []string) error { + msg := subscribeMsg{Type: "subscribe", Channel: channel, ProductIDs: productIDs} + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshal subscribe %s: %w", channel, err) + } + if err := c.c.Write(ctx, websocket.MessageText, body); err != nil { + return fmt.Errorf("write subscribe %s: %w", channel, err) + } + return nil +} + +type rawEnvelope struct { + Channel string `json:"channel"` + SequenceNum int64 `json:"sequence_num"` + Timestamp time.Time `json:"timestamp"` + Events []rawEnvelopeEv `json:"events"` +} + +type rawEnvelopeEv struct { + Type string `json:"type"` + Tickers []TickerEntry `json:"tickers,omitempty"` + CurrentTime time.Time `json:"current_time,omitempty"` +} + +var ErrFrameMalformed = errors.New("coinbase: malformed frame") + +func (c *Conn) ReadFrame(ctx context.Context) (Frame, error) { + _, msg, err := c.c.Read(ctx) + if err != nil { + return Frame{}, fmt.Errorf("read coinbase frame: %w", err) + } + return decodeFrame(msg) +} + +func decodeFrame(msg []byte) (Frame, error) { + var env rawEnvelope + if err := json.Unmarshal(msg, &env); err != nil { + return Frame{}, fmt.Errorf("%w: %v", ErrFrameMalformed, err) + } + + frame := Frame{ + SequenceNum: env.SequenceNum, + Timestamp: env.Timestamp, + } + + switch env.Channel { + case channelSubscriptions: + frame.Kind = FrameTypeSubscriptions + return frame, nil + + case channelHeartbeats: + frame.Kind = FrameTypeHeartbeats + if len(env.Events) > 0 { + frame.HeartbeatTime = env.Events[0].CurrentTime + } + return frame, nil + + case channelTicker: + for _, ev := range env.Events { + frame.Tickers = append(frame.Tickers, ev.Tickers...) + if ev.Type == "snapshot" { + frame.Kind = FrameTypeSnapshot + } + } + if frame.Kind != FrameTypeSnapshot { + frame.Kind = FrameTypeTicker + } + return frame, nil + } + + frame.Kind = FrameTypeUnknown + return frame, nil +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client_test.go new file mode 100644 index 00000000..617df843 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/client_test.go @@ -0,0 +1,190 @@ +// ©AngelaMos | 2026 +// client_test.go + +package coinbase_test + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/coinbase" +) + +type fakeServer struct { + t *testing.T + frames [][]byte + received []string + mu sync.Mutex + srv *httptest.Server +} + +func newFakeServer(t *testing.T, frames ...[]byte) *fakeServer { + t.Helper() + fs := &fakeServer{t: t, frames: frames} + fs.srv = httptest.NewServer(http.HandlerFunc(fs.handle)) + t.Cleanup(fs.srv.Close) + return fs +} + +func (fs *fakeServer) URL() string { + return "ws" + strings.TrimPrefix(fs.srv.URL, "http") +} + +func (fs *fakeServer) Received() []string { + fs.mu.Lock() + defer fs.mu.Unlock() + out := make([]string, len(fs.received)) + copy(out, fs.received) + return out +} + +func (fs *fakeServer) handle(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer conn.Close(websocket.StatusNormalClosure, "") + + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + + for i := 0; i < 2; i++ { + _, msg, err := conn.Read(ctx) + if err != nil { + return + } + fs.mu.Lock() + fs.received = append(fs.received, string(msg)) + fs.mu.Unlock() + } + + for _, frame := range fs.frames { + if err := conn.Write(ctx, websocket.MessageText, frame); err != nil { + return + } + } + + <-ctx.Done() +} + +func loadFixture(t *testing.T, name string) []byte { + t.Helper() + b, err := os.ReadFile("testdata/" + name) + require.NoError(t, err) + return b +} + +func TestClient_DialSubscribesTickerAndHeartbeats(t *testing.T) { + fs := newFakeServer(t, + loadFixture(t, "subscriptions.json"), + loadFixture(t, "ticker.json"), + ) + + d := coinbase.NewWSDialer(coinbase.DialerConfig{ + URL: fs.URL(), + ProductIDs: []string{"BTC-USD", "ETH-USD"}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + conn, err := d.Dial(ctx) + require.NoError(t, err) + defer conn.Close() + + frame, err := conn.ReadFrame(ctx) + require.NoError(t, err) + require.Equal(t, coinbase.FrameTypeSubscriptions, frame.Kind) + + frame, err = conn.ReadFrame(ctx) + require.NoError(t, err) + require.Equal(t, coinbase.FrameTypeTicker, frame.Kind) + require.Len(t, frame.Tickers, 1) + require.Equal(t, "BTC-USD", frame.Tickers[0].ProductID) + require.Equal(t, "42163.45", frame.Tickers[0].Price.String()) + + require.Eventually(t, func() bool { + recv := fs.Received() + if len(recv) < 2 { + return false + } + var subTicker, subHB struct { + Type string `json:"type"` + ProductIDs []string `json:"product_ids"` + Channel string `json:"channel"` + } + if err := json.Unmarshal([]byte(recv[0]), &subTicker); err != nil { + return false + } + if err := json.Unmarshal([]byte(recv[1]), &subHB); err != nil { + return false + } + return subTicker.Type == "subscribe" && + subTicker.Channel == "ticker" && + len(subTicker.ProductIDs) == 2 && + subHB.Type == "subscribe" && + subHB.Channel == "heartbeats" + }, time.Second, 20*time.Millisecond) +} + +func TestClient_DecodeSnapshotFrameYieldsAllTickers(t *testing.T) { + fs := newFakeServer(t, + loadFixture(t, "subscriptions.json"), + loadFixture(t, "snapshot.json"), + ) + + d := coinbase.NewWSDialer(coinbase.DialerConfig{ + URL: fs.URL(), + ProductIDs: []string{"BTC-USD", "ETH-USD"}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + conn, err := d.Dial(ctx) + require.NoError(t, err) + defer conn.Close() + + for { + frame, err := conn.ReadFrame(ctx) + require.NoError(t, err) + if frame.Kind == coinbase.FrameTypeSnapshot { + require.Len(t, frame.Tickers, 2) + return + } + } +} + +func TestClient_DecodeHeartbeatExposesCurrentTime(t *testing.T) { + fs := newFakeServer(t, + loadFixture(t, "subscriptions.json"), + loadFixture(t, "heartbeats.json"), + ) + + d := coinbase.NewWSDialer(coinbase.DialerConfig{URL: fs.URL(), ProductIDs: []string{"BTC-USD"}}) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + conn, err := d.Dial(ctx) + require.NoError(t, err) + defer conn.Close() + + for { + frame, err := conn.ReadFrame(ctx) + require.NoError(t, err) + if frame.Kind == coinbase.FrameTypeHeartbeats { + require.False(t, frame.HeartbeatTime.IsZero()) + return + } + } +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/heartbeats.json b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/heartbeats.json new file mode 100644 index 00000000..e99d33e7 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/heartbeats.json @@ -0,0 +1,9 @@ +{ + "channel": "heartbeats", + "client_id": "", + "timestamp": "2026-05-01T22:30:00.000Z", + "sequence_num": 12346, + "events": [ + {"current_time": "2026-05-01T22:30:00.000Z", "heartbeat_counter": 42} + ] +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/snapshot.json b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/snapshot.json new file mode 100644 index 00000000..73019b59 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/snapshot.json @@ -0,0 +1,15 @@ +{ + "channel": "ticker", + "client_id": "", + "timestamp": "2026-05-01T22:30:01.000Z", + "sequence_num": 1, + "events": [ + { + "type": "snapshot", + "tickers": [ + {"type": "ticker", "product_id": "BTC-USD", "price": "42163.45", "volume_24_h": "19834.9123", "time": "2026-05-01T22:30:01.000Z"}, + {"type": "ticker", "product_id": "ETH-USD", "price": "2312.75", "volume_24_h": "88123.4567", "time": "2026-05-01T22:30:01.000Z"} + ] + } + ] +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/subscriptions.json b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/subscriptions.json new file mode 100644 index 00000000..d7383fc4 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/subscriptions.json @@ -0,0 +1,9 @@ +{ + "channel": "subscriptions", + "client_id": "", + "timestamp": "2026-05-01T22:29:59.000Z", + "sequence_num": 0, + "events": [ + {"subscriptions": {"ticker": ["BTC-USD","ETH-USD"], "heartbeats": [""]}} + ] +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/ticker.json b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/ticker.json new file mode 100644 index 00000000..1c02de25 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/testdata/ticker.json @@ -0,0 +1,20 @@ +{ + "channel": "ticker", + "client_id": "", + "timestamp": "2026-05-01T22:30:00.000Z", + "sequence_num": 12345, + "events": [ + { + "type": "update", + "tickers": [ + { + "type": "ticker", + "product_id": "BTC-USD", + "price": "42163.45", + "volume_24_h": "19834.9123", + "time": "2026-05-01T22:30:00.000Z" + } + ] + } + ] +}