feat(api): WebSocket /v1/stream live tick fan-out

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)

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
 )
 

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=

api/internal/config/config.go

@@ -36,19 +36,32 @@ 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
 // `.env.example`. Returns an error only if a value is present but
 // 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

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)
+	}
+}