diff --git a/README.md b/README.md index 6ef106d..a622df5 100644 --- a/README.md +++ b/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 | diff --git a/cmd/server/main.go b/cmd/server/main.go index 6761404..7f42979 100644 --- a/cmd/server/main.go +++ b/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) diff --git a/go.mod b/go.mod index 1a52d3d..f44bcaf 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index a984215..db67e8c 100644 --- a/go.sum +++ b/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= diff --git a/internal/handlers/executions.go b/internal/handlers/executions.go index b2aa140..b1cfd15 100644 --- a/internal/handlers/executions.go +++ b/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, diff --git a/internal/handlers/handler.go b/internal/handlers/handler.go index f0e25c3..663b079 100644 --- a/internal/handlers/handler.go +++ b/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, } } diff --git a/internal/handlers/proposals.go b/internal/handlers/proposals.go index 19399a7..14d6675 100644 --- a/internal/handlers/proposals.go +++ b/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, diff --git a/internal/server/server.go b/internal/server/server.go index 85b20b9..bab0660 100644 --- a/internal/server/server.go +++ b/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) } diff --git a/internal/ws/pubsub.go b/internal/ws/pubsub.go new file mode 100644 index 0000000..aa001bc --- /dev/null +++ b/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() +} diff --git a/tmp/server b/tmp/server index 01f6c03..7c751a5 100755 Binary files a/tmp/server and b/tmp/server differ