FEAT: Issue #5 — Redis Pub/Sub Integration for Real-Time Broadcasting

README.md

@@ -317,6 +317,36 @@ docker-compose -f docker-compose.prod.yml up
 
 ---
 
+## Real-Time Broadcasting (Redis Pub/Sub)
+
+All state changes broadcast via **WebSocket + Redis pub/sub** for multi-instance support.
+
+**Architecture:**
+```
+Agent submits proposal
+    ↓
+POST /api/v1/intents/{id}/proposals
+    ├→ h.hub.Broadcast()              (local WebSocket clients)
+    └→ h.pubsub.Publish()             (Redis pub/sub channel)
+         ↓
+    Other backend instances
+    subscribe & forward to local clients
+         ↓
+    All dApp users see update in real-time
+```
+
+**Benefits:**
+- Horizontal scaling — add backend replicas, share Redis
+- No message loss — Redis is single source of truth
+- No stale data — all clients get same update
+- Real-time leaderboard updates across all intents
+
+**Channels:**
+- `intent:{intentID}` — proposals, winner selection, execution
+- `leaderboard` — agent reputation updates
+
+---
+
 ## Development Workflow
 
 ### Adding new API endpoints
@@ -405,7 +435,7 @@ Issue tracking for planned backend work:
 | #4 | Server setup (HTTP + WebSocket) | ✅ Done |
 | #2 | Database tables (models + migrations) | ✅ Done |
 | #3 | API layer (all REST endpoints) | ✅ Done |
-| #5 | Redis pub/sub integration | 🔄 Planned |
+| #5 | Redis pub/sub integration | ✅ Done |
 | #6 | Scoring engine | 🔄 Planned |
 | #7 | Execution validation service | 🔄 Planned |
 | #8 | Circle integration | 🔄 Planned |

cmd/server/main.go

@@ -31,7 +31,10 @@ func main() {
 		log.Fatalf("database migration failed: %v", err)
 	}
 
-	srv := server.New(cfg, database)
+	srv, err := server.New(cfg, database)
+	if err != nil {
+		log.Fatalf("server init failed: %v", err)
+	}
 
 	go func() {
 		log.Printf("server listening on :%s", cfg.Port)

go.mod

@@ -6,10 +6,8 @@ require (
 	github.com/go-chi/chi/v5 v5.3.0
 	github.com/gorilla/websocket v1.5.3
 	github.com/joho/godotenv v1.5.1
-)
-
-require (
 	filippo.io/edwards25519 v1.1.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/go-sql-driver/mysql v1.8.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
@@ -19,6 +17,8 @@ require (
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/lib/pq v1.12.3 // indirect
+	github.com/redis/go-redis/v9 v9.20.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
 	golang.org/x/text v0.21.0 // indirect

go.sum

@@ -1,5 +1,7 @@
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
 github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
@@ -27,9 +29,13 @@ github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwA
 github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ=
 github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0=
+github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=

internal/handlers/executions.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"time"
@@ -51,6 +52,15 @@ func (h *Handler) SelectWinner(w http.ResponseWriter, r *http.Request) {
 		"score":       winner.Score,
 	})
 
+	// Publish to Redis pub/sub
+	ctx, cancel := context.WithTimeout(context.Background(), 5*1e9)
+	defer cancel()
+	_ = h.pubsub.Publish(ctx, intentID, "winner_selected", map[string]any{
+		"proposal_id": winner.ID,
+		"agent_id":    winner.AgentID,
+		"score":       winner.Score,
+	})
+
 	respondJSON(w, http.StatusOK, map[string]any{
 		"winner_id": winner.ID,
 		"agent_id":  winner.AgentID,
@@ -165,6 +175,16 @@ func (h *Handler) SettleExecution(w http.ResponseWriter, r *http.Request) {
 		"tx_hash":      execution.TxHash,
 	})
 
+	// Publish to Redis pub/sub
+	ctx, cancel := context.WithTimeout(context.Background(), 5*1e9)
+	defer cancel()
+	_ = h.pubsub.Publish(ctx, intentID, "execution_settled", map[string]any{
+		"execution_id": execution.ID,
+		"agent_id":     execution.WinningAgentID,
+		"slippage":     execution.ActualSlippage,
+		"tx_hash":      execution.TxHash,
+	})
+
 	respondJSON(w, http.StatusOK, map[string]any{
 		"execution_id":    execution.ID,
 		"status":          execution.SettlementStatus,

internal/handlers/handler.go

@@ -11,17 +11,16 @@ import (
 )
 
 type Handler struct {
-	db  *db.DB
-	hub *ws.Hub
-	// pubsub *ws.PubSub
+	db     *db.DB
+	hub    *ws.Hub
+	pubsub *ws.PubSub
 }
 
-// redisPubSub *ws.PubSub
-func New(database *db.DB, wsHub *ws.Hub) *Handler {
+func New(database *db.DB, wsHub *ws.Hub, redisPubSub *ws.PubSub) *Handler {
 	return &Handler{
-		db:  database,
-		hub: wsHub,
-		// pubsub: redisPubSub,
+		db:     database,
+		hub:    wsHub,
+		pubsub: redisPubSub,
 	}
 }
 

internal/handlers/proposals.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 
@@ -58,6 +59,17 @@ func (h *Handler) CreateProposal(w http.ResponseWriter, r *http.Request) {
 		"quality":     proposal.ProjectedExecutionQuality,
 	})
 
+	// Publish to Redis pub/sub (multi-instance broadcast)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*1e9)
+	defer cancel()
+	_ = h.pubsub.Publish(ctx, intentID, "proposal_received", map[string]any{
+		"proposal_id": proposal.ID,
+		"agent_id":    proposal.AgentID,
+		"strategy":    proposal.StrategyType,
+		"slippage":    proposal.ProjectedSlippage,
+		"quality":     proposal.ProjectedExecutionQuality,
+	})
+
 	respondJSON(w, http.StatusCreated, ProposalResponse{
 		ID:                        proposal.ID,
 		IntentID:                  proposal.IntentID,

internal/server/server.go

@@ -26,14 +26,24 @@ type Server struct {
 	db     *db.DB
 	router *chi.Mux
 	hub    *ws.Hub
+	pubsub *ws.PubSub
 	http   *http.Server
 }
 
-func New(cfg *config.Config, database *db.DB) *Server {
+func New(cfg *config.Config, database *db.DB) (*Server, error) {
+	hub := ws.NewHub()
+
+	// Initialize Redis pub/sub
+	pubsub, err := ws.NewPubSub(cfg.RedisURL, hub)
+	if err != nil {
+		return nil, err
+	}
+
 	s := &Server{
-		cfg:  cfg,
-		db:   database,
-		hub:  ws.NewHub(),
+		cfg:    cfg,
+		db:     database,
+		hub:    hub,
+		pubsub: pubsub,
 	}
 	s.router = s.buildRouter()
 	s.http = &http.Server{
@@ -43,7 +53,7 @@ func New(cfg *config.Config, database *db.DB) *Server {
 		WriteTimeout: 15 * time.Second,
 		IdleTimeout:  60 * time.Second,
 	}
-	return s
+	return s, nil
 }
 
 func (s *Server) buildRouter() *chi.Mux {
@@ -59,7 +69,7 @@ func (s *Server) buildRouter() *chi.Mux {
 	r.Get("/ws/intents/{intentID}", s.handleWS)
 
 	// API routes
-	h := handlers.New(s.db, s.hub)
+	h := handlers.New(s.db, s.hub, s.pubsub)
 	r.Route("/api/v1", func(r chi.Router) {
 		h.Mount(r)
 	})
@@ -85,11 +95,15 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
 func (s *Server) Hub() *ws.Hub { return s.hub }
 func (s *Server) Router() *chi.Mux { return s.router }
 func (s *Server) DB() *db.DB { return s.db }
+func (s *Server) PubSub() *ws.PubSub { return s.pubsub }
 
 func (s *Server) Start() error {
 	return s.http.ListenAndServe()
 }
 
 func (s *Server) Shutdown(ctx context.Context) error {
+	if err := s.pubsub.Close(); err != nil {
+		return err
+	}
 	return s.http.Shutdown(ctx)
 }

internal/ws/pubsub.go

@@ -0,0 +1,127 @@
+package ws
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+
+	"github.com/redis/go-redis/v9"
+)
+
+type PubSub struct {
+	client *redis.Client
+	hub    *Hub
+	subs   map[string]*redis.PubSub
+}
+
+func NewPubSub(redisURL string, hub *Hub) (*PubSub, error) {
+	opt, err := redis.ParseURL(redisURL)
+	if err != nil {
+		return nil, fmt.Errorf("invalid redis url: %w", err)
+	}
+
+	client := redis.NewClient(opt)
+
+	// Test connection
+	ctx, cancel := context.WithTimeout(context.Background(), 5*1e9) // 5 seconds
+	defer cancel()
+	if err := client.Ping(ctx).Err(); err != nil {
+		return nil, fmt.Errorf("redis connection failed: %w", err)
+	}
+
+	ps := &PubSub{
+		client: client,
+		hub:    hub,
+		subs:   make(map[string]*redis.PubSub),
+	}
+
+	return ps, nil
+}
+
+// Subscribe listens for messages on a channel and forwards to WebSocket hub
+func (ps *PubSub) Subscribe(ctx context.Context, channel string) {
+	sub := ps.client.Subscribe(ctx, channel)
+	ps.subs[channel] = sub
+
+	go func() {
+		ch := sub.Channel()
+		for msg := range ch {
+			// Parse message
+			var payload map[string]any
+			if err := json.Unmarshal([]byte(msg.Payload), &payload); err != nil {
+				log.Printf("failed to unmarshal pubsub message: %v", err)
+				continue
+			}
+
+			// Extract intent ID and message type
+			msgType, ok := payload["type"].(string)
+			if !ok {
+				continue
+			}
+
+			data, ok := payload["payload"]
+			if !ok {
+				continue
+			}
+
+			intentID, ok := payload["intent_id"].(string)
+			if !ok {
+				continue
+			}
+
+			// Broadcast to local WebSocket subscribers
+			ps.hub.Broadcast(intentID, msgType, data)
+		}
+	}()
+}
+
+// Publish sends a message to Redis pub/sub
+func (ps *PubSub) Publish(ctx context.Context, intentID, msgType string, payload any) error {
+	message := map[string]any{
+		"type":      msgType,
+		"payload":   payload,
+		"intent_id": intentID,
+	}
+
+	data, err := json.Marshal(message)
+	if err != nil {
+		return fmt.Errorf("failed to marshal message: %w", err)
+	}
+
+	channel := fmt.Sprintf("intent:%s", intentID)
+	if err := ps.client.Publish(ctx, channel, string(data)).Err(); err != nil {
+		return fmt.Errorf("redis publish failed: %w", err)
+	}
+
+	return nil
+}
+
+// PublishLeaderboardUpdate broadcasts leaderboard changes to all clients
+func (ps *PubSub) PublishLeaderboardUpdate(ctx context.Context, payload any) error {
+	message := map[string]any{
+		"type":    "leaderboard_update",
+		"payload": payload,
+	}
+
+	data, err := json.Marshal(message)
+	if err != nil {
+		return fmt.Errorf("failed to marshal message: %w", err)
+	}
+
+	if err := ps.client.Publish(ctx, "leaderboard", string(data)).Err(); err != nil {
+		return fmt.Errorf("redis publish failed: %w", err)
+	}
+
+	return nil
+}
+
+// Close closes all subscriptions and the Redis client
+func (ps *PubSub) Close() error {
+	for _, sub := range ps.subs {
+		if err := sub.Close(); err != nil {
+			return err
+		}
+	}
+	return ps.client.Close()
+}