feat(monitor/collectors/coinbase): WS dialer + frame decoder for ticker/heartbeats/snapshot
This commit is contained in:
CarterPerez-dev
parent
ae5a83c2f8
commit
10862314f0
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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}
|
||||
]
|
||||
}
|
||||
|
|
@ -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"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -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": [""]}}
|
||||
]
|
||||
}
|
||||
|
|
@ -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"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
Loading…
Reference in New Issue