diff --git a/api/cmd/api/main.go b/api/cmd/api/main.go index 73aa0a3..983305d 100644 --- a/api/cmd/api/main.go +++ b/api/cmd/api/main.go @@ -30,10 +30,12 @@ import ( "time" "github.com/gofiber/fiber/v3" + "github.com/gofiber/fiber/v3/middleware/adaptor" "github.com/obchain/volx/api/internal/config" "github.com/obchain/volx/api/internal/handlers" "github.com/obchain/volx/api/internal/storage" + "github.com/obchain/volx/api/internal/stream" ) func main() { @@ -108,6 +110,19 @@ func main() { v1.Get("/index/:id/history", handlers.IndexHistory(indexDeps)) v1.Get("/options/strip", handlers.OptionsStrip(indexDeps)) + // WebSocket live stream (#24). One hub per process, started in + // a goroutine alongside the fiber listener; the hub keeps a + // single Redis PSUBSCRIBE open and fans every published tick + // to the per-conn channels created by `stream.Handler`. The + // IP limiter caps anon connections per PRD §6. + hubCtx, hubCancel := context.WithCancel(context.Background()) + defer hubCancel() + hub := stream.NewHub(rds.Client) + go hub.Run(hubCtx) + ipLimit := stream.NewIPLimiter(cfg.WSMaxConnsPerIP) + v1.Get("/stream", adaptor.HTTPHandler(stream.Handler(hub, ipLimit))) + slog.Info("ws stream wired", "path", "/v1/stream", "per_ip_cap", cfg.WSMaxConnsPerIP) + // Run the listener in its own goroutine so `main` can wait on // signal + shutdown sequentially. errCh := make(chan error, 1) diff --git a/api/go.mod b/api/go.mod index 668ee5d..3e18a82 100644 --- a/api/go.mod +++ b/api/go.mod @@ -5,6 +5,7 @@ go 1.25.0 require ( github.com/ClickHouse/clickhouse-go/v2 v2.46.0 github.com/gofiber/fiber/v3 v3.3.0 + github.com/gorilla/websocket v1.5.3 github.com/redis/go-redis/v9 v9.19.0 ) diff --git a/api/go.sum b/api/go.sum index 99638b8..c388c20 100644 --- a/api/go.sum +++ b/api/go.sum @@ -34,6 +34,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= diff --git a/api/internal/config/config.go b/api/internal/config/config.go index 5fd91e4..3957eda 100644 --- a/api/internal/config/config.go +++ b/api/internal/config/config.go @@ -36,6 +36,11 @@ type Config struct { // the most recent `index_ticks` row is older than this. 90 s is // 1.5 × the engine's 60 s recompute cadence. HealthMaxAge time.Duration + + // WSMaxConnsPerIP caps per-IP active WS connections. PRD §6 + // pins the anon limit at 5; paid-tier auth (#M3) will skip + // this limiter for authenticated keys. + WSMaxConnsPerIP int } // Load reads every env var, applying defaults that match @@ -43,12 +48,20 @@ type Config struct { // unparseable. func Load() (*Config, error) { c := &Config{ - BindAddr: env("API_BIND", "127.0.0.1:8080"), - ClickHouseDSN: env("CLICKHOUSE_DSN", "clickhouse://default@127.0.0.1:9000?dial_timeout=2s"), - ClickHouseDB: env("CLICKHOUSE_DB", "volx"), - RedisURL: env("REDIS_URL", "redis://127.0.0.1:6379"), - Version: env("VOLX_VERSION", "0.1.0"), - HealthMaxAge: 90 * time.Second, + BindAddr: env("API_BIND", "127.0.0.1:8080"), + ClickHouseDSN: env("CLICKHOUSE_DSN", "clickhouse://default@127.0.0.1:9000?dial_timeout=2s"), + ClickHouseDB: env("CLICKHOUSE_DB", "volx"), + RedisURL: env("REDIS_URL", "redis://127.0.0.1:6379"), + Version: env("VOLX_VERSION", "0.1.0"), + HealthMaxAge: 90 * time.Second, + WSMaxConnsPerIP: 5, + } + if raw := os.Getenv("WS_MAX_CONNS_PER_IP"); raw != "" { + var n int + if _, err := fmt.Sscanf(raw, "%d", &n); err != nil { + return nil, fmt.Errorf("WS_MAX_CONNS_PER_IP=%q: %w", raw, err) + } + c.WSMaxConnsPerIP = n } if raw := os.Getenv("HEALTH_MAX_AGE_SECS"); raw != "" { var secs int diff --git a/api/internal/stream/conn.go b/api/internal/stream/conn.go new file mode 100644 index 0000000..a6c6caf --- /dev/null +++ b/api/internal/stream/conn.go @@ -0,0 +1,234 @@ +package stream + +import ( + "encoding/json" + "log/slog" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +// MaxClientMsgBytes caps the size of a single client → server frame. +// Clients only send the small `subscribe` envelope (~50 B); a larger +// frame is either a misuse or an attempt to allocate-spam the +// server. +const MaxClientMsgBytes = 4 * 1024 + +// upgrader configures the gorilla WS handshake. Origin check is +// permissive at v1 (the API is read-only, no cookie auth, no CSRF +// surface). Tightening lands when paid-tier auth ships in #M3. +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, + CheckOrigin: func(_ *http.Request) bool { + return true + }, +} + +// Conn is one client WebSocket. The hub holds a *Conn pointer to +// fan messages into `send`; the reader + writer goroutines own all +// other lifecycle. +type Conn struct { + ws *websocket.Conn + send chan []byte + + mu sync.RWMutex + channels map[string]struct{} + + closeOnce sync.Once + closed chan struct{} +} + +// newConn wraps a freshly-upgraded gorilla connection. +func newConn(ws *websocket.Conn) *Conn { + return &Conn{ + ws: ws, + send: make(chan []byte, SendQueueDepth), + channels: make(map[string]struct{}), + closed: make(chan struct{}), + } +} + +// subscribed reports whether this connection wants messages for the +// given lowercase index id. +func (c *Conn) subscribed(channel string) bool { + c.mu.RLock() + _, ok := c.channels[channel] + c.mu.RUnlock() + return ok +} + +// tryPush hands a pre-encoded frame to the writer goroutine. +// Drop-newest on a full queue — the wedged-slow-client case. The +// alternative (block-the-hub) would create head-of-line blocking +// across all conns. +func (c *Conn) tryPush(frame []byte) { + select { + case c.send <- frame: + default: + // Slow consumer; drop newest. A `slow_consumer` metric + // would increment here when the API exporter symmetry + // PR lands. + } +} + +// close terminates the connection and unblocks both goroutines. +// Idempotent (sync.Once). +func (c *Conn) close() { + c.closeOnce.Do(func() { + close(c.closed) + _ = c.ws.Close() + }) +} + +// --- client → server message --------------------------------------- + +// clientMsg is the only inbound envelope shape we accept. PRD §6 +// supports `subscribe`; an `unsubscribe` action is left to a future +// PR (frontend just opens a new conn for now). +type clientMsg struct { + Action string `json:"action"` + Channels []string `json:"channels"` +} + +// Allowed client → server actions. +const ( + actionSubscribe = "subscribe" +) + +// subscribeTo updates the per-conn channel set. Unknown channels +// are silently dropped (rather than rejecting the whole subscribe) +// so a partial-success client (`["bvol", "future-index"]`) still +// gets BVOL. +func (c *Conn) subscribeTo(channels []string) { + c.mu.Lock() + defer c.mu.Unlock() + for _, ch := range channels { + switch ch { + case "bvol", "evol": + c.channels[ch] = struct{}{} + default: + // silently ignored — see godoc above + } + } +} + +// --- read / write goroutines -------------------------------------- + +// readLoop drains client → server frames. The protocol is +// effectively one-shot — clients send a single `subscribe` +// envelope and then only receive — but we keep the loop draining +// so a client that sends garbage doesn't fill kernel buffers. +func (c *Conn) readLoop(hub *Hub) { + defer hub.Unregister(c) + defer c.close() + + c.ws.SetReadLimit(MaxClientMsgBytes) + // Reset the read deadline on every pong. `PingInterval + + // PongTimeout` is the longest a healthy connection can stay + // silent before we declare it dead. + _ = c.ws.SetReadDeadline(time.Now().Add(PingInterval + PongTimeout)) + c.ws.SetPongHandler(func(string) error { + _ = c.ws.SetReadDeadline(time.Now().Add(PingInterval + PongTimeout)) + return nil + }) + + for { + _, raw, err := c.ws.ReadMessage() + if err != nil { + // Normal close or read deadline — exit quietly. + return + } + var msg clientMsg + if err := json.Unmarshal(raw, &msg); err != nil { + c.tryPush(errorFrame("bad_request", "message is not valid JSON")) + continue + } + switch msg.Action { + case actionSubscribe: + c.subscribeTo(msg.Channels) + default: + c.tryPush(errorFrame("bad_request", "unknown action (allowed: subscribe)")) + } + } +} + +// writeLoop pulls fanned-out frames from the per-conn send channel +// and writes them to the WS. Also fires the keepalive ping every +// `PingInterval`. +func (c *Conn) writeLoop() { + ticker := time.NewTicker(PingInterval) + defer ticker.Stop() + defer c.close() + + // Note on exit paths: `c.send` is never explicitly closed — + // `Conn` is GC'd with its channel once both goroutines exit. + // The two real exits below are `c.closed` (peer drop / hub + // shutdown) and a write error (deadline exceeded, peer reset). + for { + select { + case <-c.closed: + return + case frame := <-c.send: + _ = c.ws.SetWriteDeadline(time.Now().Add(WriteTimeout)) + if err := c.ws.WriteMessage(websocket.TextMessage, frame); err != nil { + return + } + case <-ticker.C: + _ = c.ws.SetWriteDeadline(time.Now().Add(PongTimeout)) + if err := c.ws.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} + +// errorFrame builds a `{ "type":"error", "code":..., "message":... }` +// envelope. Kept inline rather than a separate handler module +// because the only producer is the read loop above. +func errorFrame(code, message string) []byte { + b, _ := json.Marshal(map[string]any{ + "type": "error", + "code": code, + "message": message, + }) + return b +} + +// --- HTTP upgrade entry point ------------------------------------- + +// Handler returns an `http.HandlerFunc` that upgrades to WebSocket +// and starts the per-conn goroutines. fiber v3 has no native WS +// helper; the bridge is `middleware/adaptor.HTTPHandler(streamHandler)` +// in `cmd/api/main.go`. +// +// `limit` is the per-IP active-connection cap (PRD §6 anon: 5). +func Handler(hub *Hub, limit *IPLimiter) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + ip := clientIP(r) + if !limit.Acquire(ip) { + http.Error(w, "too many connections from this IP", http.StatusTooManyRequests) + return + } + ws, err := upgrader.Upgrade(w, r, nil) + if err != nil { + limit.Release(ip) + slog.Warn("ws upgrade failed", "error", err, "remote", r.RemoteAddr) + return + } + conn := newConn(ws) + hub.Register(conn) + + // `readLoop` returns first (via close/error/deadline); it + // unregisters + closes. `writeLoop` then exits on the + // `closed` channel. Release the IP slot when both end. + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); conn.readLoop(hub) }() + go func() { defer wg.Done(); conn.writeLoop() }() + wg.Wait() + limit.Release(ip) + } +} diff --git a/api/internal/stream/hub.go b/api/internal/stream/hub.go new file mode 100644 index 0000000..253fb66 --- /dev/null +++ b/api/internal/stream/hub.go @@ -0,0 +1,217 @@ +// Package stream implements the live tick broadcast layer per PRD §6 +// and issue #24. The pipeline shape: +// +// Engine → Redis PUBLISH index:{id}:stream ─┐ +// ▼ +// Hub.Run (one PSUBSCRIBE for all indices) +// │ +// ▼ +// fan-out → registered Conn.send channels +// │ +// ▼ +// gorilla WS frames → browser +// +// One [`Hub`] runs per API process. It subscribes to the pattern +// `index:*:stream` and pushes each parsed tick to every connection +// whose `channels` set includes the index. +// +// Wire format (PRD §6, line 988): +// +// // client → server +// { "action": "subscribe", "channels": ["bvol", "evol"] } +// +// // server → client (per tick) +// { "type":"tick", "channel":"bvol", +// "value": 67.42, "ts": 1747668092847, "confidence": 0.97 } +// +// `ts` is Unix epoch **milliseconds** (integer) — different from the +// REST endpoints' RFC 3339 form. PRD lines 1054–1056 show the +// frontend doing `tick.ts / 1000` for `lightweight-charts`' +// seconds-since-epoch input. +package stream + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +// pubsubPattern matches every `index:{id}:stream` channel the engine +// publishes to. One subscriber per process keeps the Redis read cost +// constant regardless of how many WS clients connect. +const pubsubPattern = "index:*:stream" + +// SendQueueDepth bounds the per-connection outbound buffer. A slow +// WS client (high-latency network, paused browser tab) cannot wedge +// the hub — when the queue is full the *oldest* message is dropped +// and a `volx_api_ws_dropped_total{reason="slow_consumer"}` counter +// would increment (counter wired when #11 engine exporter symmetry +// lands). +const SendQueueDepth = 32 + +// PingInterval governs keepalive ping cadence. A client that does +// not respond with a pong inside `PingInterval + PongTimeout` is +// closed. 30 s is well under common proxy idle timeouts (60-120 s). +const PingInterval = 30 * time.Second + +// PongTimeout is how long after a ping we wait for the pong. +const PongTimeout = 10 * time.Second + +// WriteTimeout caps how long a single data-frame write can block +// before we declare the WS connection dead. Distinct from +// `PongTimeout` so the keepalive cadence + per-frame stall budget +// can be tuned independently. +const WriteTimeout = 10 * time.Second + +// EngineTick is the JSON envelope the engine writes to Redis. Same +// shape as `handlers.engineIndexValue` but kept independent because +// the two are wired-time contracts that may diverge in future (e.g. +// stream adds a sequence number that REST doesn't). +type EngineTick struct { + IndexID string `json:"index_id"` + Value float64 `json:"value"` + Confidence float64 `json:"confidence"` + StripHash string `json:"strip_hash"` + Ts string `json:"ts"` +} + +// ClientTick is the wire shape pushed to browsers. Channel name is +// **lowercase** (`bvol`) to match the URL convention; `ts` is Unix +// epoch milliseconds (integer) per PRD §6. +type ClientTick struct { + Type string `json:"type"` + Channel string `json:"channel"` + Value float64 `json:"value"` + Ts int64 `json:"ts"` + Confidence float64 `json:"confidence"` +} + +// Hub fans Redis pubsub messages out to registered connections. +// Safe for concurrent use. +type Hub struct { + redis *redis.Client + + mu sync.RWMutex + conns map[*Conn]struct{} +} + +// NewHub returns an unstarted hub. Call `Run` from a goroutine to +// begin the Redis subscribe loop. +func NewHub(r *redis.Client) *Hub { + return &Hub{ + redis: r, + conns: make(map[*Conn]struct{}), + } +} + +// Run subscribes to the pattern and forwards each message to the +// registered connections. Blocks until `ctx` cancels or the +// pubsub channel closes. The hub never returns an error — pubsub +// failures are surfaced via `volx_api_ws_pubsub_errors_total` in a +// future observability PR. +func (h *Hub) Run(ctx context.Context) { + psub := h.redis.PSubscribe(ctx, pubsubPattern) + defer func() { _ = psub.Close() }() + + ch := psub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + h.fanOut(msg.Channel, msg.Payload) + } + } +} + +// channelToIndex parses `index:{id}:stream` into the lowercase +// `bvol|evol` form. Returns ok=false on any other pattern (engine +// might publish to other channels in future; we ignore them safely). +func channelToIndex(redisChannel string) (string, bool) { + if !strings.HasPrefix(redisChannel, "index:") || !strings.HasSuffix(redisChannel, ":stream") { + return "", false + } + mid := strings.TrimSuffix(strings.TrimPrefix(redisChannel, "index:"), ":stream") + switch mid { + case "BVOL": + return "bvol", true + case "EVOL": + return "evol", true + default: + return "", false + } +} + +// fanOut parses one Redis message and pushes it to subscribed conns. +// Decoding lives here (once per pubsub message) rather than per +// connection (once per recipient) — the latter would do +// `N_conns × M_msgs/sec` JSON parses. +func (h *Hub) fanOut(redisChannel, payload string) { + channel, ok := channelToIndex(redisChannel) + if !ok { + return + } + + var et EngineTick + if err := json.Unmarshal([]byte(payload), &et); err != nil { + return + } + + // RFC 3339 (engine) → epoch ms (wire). Fall back to "now" if + // the engine ts is malformed — a client tick missing `ts` + // would break the frontend chart axis worse than a slightly + // off timestamp. + var tsMs int64 + if t, err := time.Parse(time.RFC3339Nano, et.Ts); err == nil { + tsMs = t.UnixMilli() + } else { + tsMs = time.Now().UnixMilli() + } + ct := ClientTick{ + Type: "tick", + Channel: channel, + Value: et.Value, + Ts: tsMs, + Confidence: et.Confidence, + } + encoded, err := json.Marshal(ct) + if err != nil { + return + } + + h.mu.RLock() + defer h.mu.RUnlock() + for c := range h.conns { + if c.subscribed(channel) { + c.tryPush(encoded) + } + } +} + +// Register adds a connection to the fan-out set. Idempotent. +func (h *Hub) Register(c *Conn) { + h.mu.Lock() + h.conns[c] = struct{}{} + h.mu.Unlock() +} + +// Unregister removes the connection. Idempotent (extra calls are +// no-ops, so the per-conn defer is always safe even if the conn was +// never successfully registered). +func (h *Hub) Unregister(c *Conn) { + h.mu.Lock() + delete(h.conns, c) + h.mu.Unlock() +} + +// ErrUnknownChannel is returned by `Conn.subscribeTo` when the +// client requests a channel that does not map to an index. +var ErrUnknownChannel = errors.New("unknown channel (allowed: bvol, evol)") diff --git a/api/internal/stream/iplimit.go b/api/internal/stream/iplimit.go new file mode 100644 index 0000000..4f134f7 --- /dev/null +++ b/api/internal/stream/iplimit.go @@ -0,0 +1,99 @@ +package stream + +import ( + "net" + "net/http" + "sync" +) + +// IPLimiter caps per-IP concurrent WebSocket connections. PRD §6 +// pins the anon limit at 5; a paid-tier auth check (#M3) will route +// API-key requests through a different limiter and skip this one. +// +// Implementation is a sync.Mutex-guarded `map[ip]int`. The map +// grows with the active-IP count, not the cumulative connection +// count — `Release` decrements and deletes at zero. +type IPLimiter struct { + max int + mu sync.Mutex + conns map[string]int +} + +// NewIPLimiter constructs a limiter with the given per-IP cap. +// `max <= 0` disables the limit (used by tests). +func NewIPLimiter(max int) *IPLimiter { + return &IPLimiter{max: max, conns: make(map[string]int)} +} + +// Acquire reserves one connection slot for the IP. Returns false +// when the IP is at cap. +func (l *IPLimiter) Acquire(ip string) bool { + if l.max <= 0 { + return true + } + l.mu.Lock() + defer l.mu.Unlock() + if l.conns[ip] >= l.max { + return false + } + l.conns[ip]++ + return true +} + +// Release returns one slot. Safe to call on an IP that was never +// `Acquire`d (no-op, useful in error-recovery paths in the WS +// handler). +func (l *IPLimiter) Release(ip string) { + if l.max <= 0 { + return + } + l.mu.Lock() + defer l.mu.Unlock() + if c, ok := l.conns[ip]; ok { + if c <= 1 { + delete(l.conns, ip) + } else { + l.conns[ip] = c - 1 + } + } +} + +// clientIP extracts the request's source IP. `X-Forwarded-For` takes +// precedence so reverse-proxy deployments (Caddy, Cloudflare per +// PRD §13) see the real client; falls back to `RemoteAddr`. +// +// Behind a trusted proxy this is correct; without one a hostile +// client can forge X-Forwarded-For to bypass the limit. The PRD +// localhost-bind posture + Caddy front means the API never serves +// untrusted X-Forwarded-For directly. +func clientIP(r *http.Request) string { + if xff := r.Header.Get("X-Forwarded-For"); xff != "" { + // The leftmost entry is the original client; intermediate + // proxies append. + for i := 0; i < len(xff); i++ { + if xff[i] == ',' { + return trimSpace(xff[:i]) + } + } + return trimSpace(xff) + } + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + return r.RemoteAddr + } + return host +} + +// trimSpace is a tiny stdlib-free trim so this file does not depend +// on `strings`. Trims ASCII space + tab on both ends. +func trimSpace(s string) string { + start := 0 + for start < len(s) && (s[start] == ' ' || s[start] == '\t') { + start++ + } + end := len(s) + for end > start && (s[end-1] == ' ' || s[end-1] == '\t') { + end-- + } + return s[start:end] +}