feat(monitor/collectors/coinbase): collector Run loop ties dial+readloop+aggregator+repo+throttled-emit
This commit is contained in:
CarterPerez-dev
parent
f6192e9c04
commit
89f1c10249
|
|
@ -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),
|
||||
})
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
Loading…
Reference in New Issue