From 70d39b87d16efd6363c4968144f883eb7a9887a4 Mon Sep 17 00:00:00 2001 From: Dex Date: Wed, 20 May 2026 11:02:21 +0000 Subject: [PATCH] CUB-203: add WebSocket client scaffold for OpenClaw gateway v3 --- go-backend/cmd/server/main.go | 9 + go-backend/go.mod | 1 + go-backend/go.sum | 2 + go-backend/internal/config/config.go | 16 +- go-backend/internal/gateway/wsclient.go | 423 ++++++++++++++++++++++++ 5 files changed, 445 insertions(+), 6 deletions(-) create mode 100644 go-backend/internal/gateway/wsclient.go diff --git a/go-backend/cmd/server/main.go b/go-backend/cmd/server/main.go index dc760b4..76b6a2b 100644 --- a/go-backend/cmd/server/main.go +++ b/go-backend/cmd/server/main.go @@ -69,11 +69,20 @@ func main() { PollInterval: cfg.GatewayPollInterval, }, agentRepo, broker) + // ── WebSocket client (connects to OpenClaw gateway WS v3) ───────────── + wsClient := gateway.NewWSClient(gateway.WSConfig{ + URL: cfg.WSGatewayURL, + AuthToken: cfg.WSGatewayToken, + }, agentRepo, broker, logger) + ctx, cancel := context.WithCancel(context.Background()) defer cancel() go gwClient.Start(ctx) + // Start WS client in background; logs connection status + go wsClient.Start(ctx) + // ── Server ───────────────────────────────────────────────────────────── srv := &http.Server{ Addr: fmt.Sprintf(":%d", cfg.Port), diff --git a/go-backend/go.mod b/go-backend/go.mod index 4b9db15..91bad3e 100644 --- a/go-backend/go.mod +++ b/go-backend/go.mod @@ -7,6 +7,7 @@ require ( github.com/go-chi/cors v1.2.1 github.com/go-playground/validator/v10 v10.24.0 github.com/google/uuid v1.6.0 + github.com/gorilla/websocket v1.5.3 github.com/jackc/pgx/v5 v5.7.2 ) diff --git a/go-backend/go.sum b/go-backend/go.sum index f4041e3..448723d 100644 --- a/go-backend/go.sum +++ b/go-backend/go.sum @@ -17,6 +17,8 @@ github.com/go-playground/validator/v10 v10.24.0 h1:KHQckvo8G6hlWnrPX4NJJ+aBfWNAE github.com/go-playground/validator/v10 v10.24.0/go.mod h1:GGzBIJMuE98Ic/kJsBXbz1x/7cByt++cQ+YOuDM5wus= 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/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= diff --git a/go-backend/internal/config/config.go b/go-backend/internal/config/config.go index fe6c9f6..3098ba8 100644 --- a/go-backend/internal/config/config.go +++ b/go-backend/internal/config/config.go @@ -10,13 +10,15 @@ import ( // Config holds all application configuration. type Config struct { - Port int - DatabaseURL string - CORSOrigin string - LogLevel string - Environment string - GatewayURL string + Port int + DatabaseURL string + CORSOrigin string + LogLevel string + Environment string + GatewayURL string GatewayPollInterval time.Duration + WSGatewayURL string + WSGatewayToken string } // Load reads configuration from environment variables, applying defaults where @@ -30,6 +32,8 @@ func Load() *Config { Environment: getEnv("ENVIRONMENT", "development"), GatewayURL: getEnv("GATEWAY_URL", "http://localhost:18789/api/agents"), GatewayPollInterval: getEnvDuration("GATEWAY_POLL_INTERVAL", 5*time.Second), + WSGatewayURL: getEnv("WS_GATEWAY_URL", "ws://localhost:18789/"), + WSGatewayToken: getEnv("OPENCLAW_GATEWAY_TOKEN", ""), } } diff --git a/go-backend/internal/gateway/wsclient.go b/go-backend/internal/gateway/wsclient.go new file mode 100644 index 0000000..1e64775 --- /dev/null +++ b/go-backend/internal/gateway/wsclient.go @@ -0,0 +1,423 @@ +// Package gateway provides WebSocket client integration with the OpenClaw +// gateway using WS protocol v3. The WSClient handles connection, handshake, +// frame routing, request/response correlation, and automatic reconnection +// with exponential backoff. +package gateway + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "sync" + "time" + + "code.cubecraftcreations.com/CubeCraft-Creations/Control-Center/go-backend/internal/handler" + "code.cubecraftcreations.com/CubeCraft-Creations/Control-Center/go-backend/internal/repository" + + "github.com/gorilla/websocket" + "github.com/google/uuid" +) + +// WSConfig holds WebSocket client configuration, typically loaded from +// environment variables. AuthToken must be set to a valid OpenClaw gateway +// operator token. +type WSConfig struct { + URL string // e.g. "ws://host.docker.internal:18789/" + AuthToken string // from OPENCLAW_GATEWAY_TOKEN +} + +// DefaultWSConfig returns sensible defaults for local development. +func DefaultWSConfig() WSConfig { + return WSConfig{ + URL: "ws://localhost:18789/", + AuthToken: "", + } +} + +// eventHandler is a callback invoked when a named event arrives from the +// gateway. +type eventHandler func(json.RawMessage) + +// WSClient connects to the OpenClaw gateway over WebSocket, completes the +// v3 handshake, routes incoming frames, and automatically reconnects on +// disconnect with exponential backoff. +type WSClient struct { + config WSConfig + conn *websocket.Conn + connMu sync.Mutex // protects conn for writes + pending map[string]chan<- json.RawMessage + mu sync.Mutex // protects pending and handlers + agents repository.AgentRepo + broker *handler.Broker + logger *slog.Logger + + handlers map[string][]eventHandler +} + +// NewWSClient returns a WSClient wired to the given repository and broker. +func NewWSClient(cfg WSConfig, agents repository.AgentRepo, broker *handler.Broker, logger *slog.Logger) *WSClient { + if logger == nil { + logger = slog.Default() + } + return &WSClient{ + config: cfg, + pending: make(map[string]chan<- json.RawMessage), + agents: agents, + broker: broker, + logger: logger, + handlers: make(map[string][]eventHandler), + } +} + +// OnEvent registers a handler for the given event name. Handlers are called +// when an incoming frame with type "event" and matching event name is +// received. This is safe to call before Start. +func (c *WSClient) OnEvent(event string, handler func(json.RawMessage)) { + c.mu.Lock() + defer c.mu.Unlock() + c.handlers[event] = append(c.handlers[event], handler) +} + +// ── Frame types ────────────────────────────────────────────────────────── + +// wsFrame represents a generic WebSocket frame in the OpenClaw v3 protocol. +type wsFrame struct { + Type string `json:"type"` // "req", "res", "event" + ID string `json:"id,omitempty"` // request/response correlation + Method string `json:"method,omitempty"` // method name (req frames) + Event string `json:"event,omitempty"` // event name (event frames) + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *wsError `json:"error,omitempty"` +} + +// wsError represents an error in a response frame. +type wsError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// connectRequest builds the initial connect handshake payload. +type connectRequest struct { + MinProtocol int `json:"minProtocol"` + MaxProtocol int `json:"maxProtocol"` + Client connectClientInfo `json:"client"` + Role string `json:"role"` + Scopes []string `json:"scopes"` + Auth connectAuth `json:"auth"` +} + +type connectClientInfo struct { + ID string `json:"id"` + Version string `json:"version"` + Platform string `json:"platform"` + Mode string `json:"mode"` +} + +type connectAuth struct { + Token string `json:"token"` +} + +// helloOKResponse represents the expected response to a successful connect. +type helloOKResponse struct { + ConnID string `json:"connId"` + Features struct { + Methods []string `json:"methods"` + Events []string `json:"events"` + } `json:"features"` +} + +// ── Start loop ─────────────────────────────────────────────────────────── + +// Start connects to the gateway, completes the handshake, and begins the +// read loop. On disconnect it reconnects with exponential backoff. On +// ctx cancellation it performs a clean shutdown. +func (c *WSClient) Start(ctx context.Context) { + backoff := 1 * time.Second + maxBackoff := 30 * time.Second + + for { + err := c.connectAndRun(ctx) + if err != nil { + if ctx.Err() != nil { + c.logger.Info("ws client stopped (context cancelled)") + return + } + c.logger.Warn("ws client disconnected, reconnecting", + "error", err, + "backoff", backoff) + } + + select { + case <-ctx.Done(): + c.logger.Info("ws client stopped during backoff (context cancelled)") + return + case <-time.After(backoff): + // Exponential backoff: 1s, 2s, 4s, 8s, 16s, max 30s + backoff = backoff * 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } + } +} + +// connectAndRun dials the gateway, completes the handshake, and runs the +// read loop until an error occurs or ctx is cancelled. +func (c *WSClient) connectAndRun(ctx context.Context) error { + c.logger.Info("ws client connecting", "url", c.config.URL) + + dialer := websocket.Dialer{ + HandshakeTimeout: 10 * time.Second, + } + + conn, _, err := dialer.DialContext(ctx, c.config.URL, nil) + if err != nil { + return fmt.Errorf("dial failed: %w", err) + } + + c.connMu.Lock() + c.conn = conn + c.connMu.Unlock() + + // Reset backoff on successful connect + defer func() { + conn.Close() + }() + + // Step 1: Read the connect.challenge frame + if err := c.readChallenge(conn); err != nil { + return fmt.Errorf("handshake challenge: %w", err) + } + + // Step 2: Send connect request + helloOK, err := c.sendConnect(conn) + if err != nil { + return fmt.Errorf("handshake connect: %w", err) + } + + c.logger.Info("ws client handshake complete", + "connId", helloOK.ConnID, + "methods", helloOK.Features.Methods, + "events", helloOK.Features.Events) + + // Step 3: Read loop + return c.readLoop(ctx, conn) +} + +// readChallenge reads the first frame from the gateway, which must be a +// connect.challenge event. +func (c *WSClient) readChallenge(conn *websocket.Conn) error { + var frame wsFrame + if err := conn.ReadJSON(&frame); err != nil { + return fmt.Errorf("read challenge: %w", err) + } + + if frame.Type != "event" || frame.Event != "connect.challenge" { + return fmt.Errorf("expected connect.challenge, got type=%s event=%s", frame.Type, frame.Event) + } + + c.logger.Debug("received connect.challenge", "params", string(frame.Params)) + return nil +} + +// sendConnect sends the connect request and waits for the hello-ok response. +func (c *WSClient) sendConnect(conn *websocket.Conn) (*helloOKResponse, error) { + reqID := uuid.New().String() + params := connectRequest{ + MinProtocol: 3, + MaxProtocol: 3, + Client: connectClientInfo{ + ID: "control-center", + Version: "1.0", + Platform: "server", + Mode: "operator", + }, + Role: "operator", + Scopes: []string{"operator.read"}, + Auth: connectAuth{ + Token: c.config.AuthToken, + }, + } + + paramsJSON, err := json.Marshal(params) + if err != nil { + return nil, fmt.Errorf("marshal connect params: %w", err) + } + + reqFrame := wsFrame{ + Type: "req", + ID: reqID, + Method: "connect", + Params: paramsJSON, + } + + if err := conn.WriteJSON(reqFrame); err != nil { + return nil, fmt.Errorf("write connect request: %w", err) + } + + // Read response + var resFrame wsFrame + if err := conn.ReadJSON(&resFrame); err != nil { + return nil, fmt.Errorf("read connect response: %w", err) + } + + if resFrame.Error != nil { + return nil, fmt.Errorf("connect rejected: code=%d msg=%s", resFrame.Error.Code, resFrame.Error.Message) + } + + if resFrame.ID != reqID { + return nil, fmt.Errorf("response id mismatch: expected %s, got %s", reqID, resFrame.ID) + } + + // Check for hello-ok method in the result + // The gateway responds with method "hello-ok" on success + var helloOK helloOKResponse + if err := json.Unmarshal(resFrame.Result, &helloOK); err != nil { + return nil, fmt.Errorf("parse hello-ok: %w", err) + } + + return &helloOK, nil +} + +// readLoop continuously reads frames from the connection and routes them. +// It returns on read error or context cancellation. +func (c *WSClient) readLoop(ctx context.Context, conn *websocket.Conn) error { + for { + select { + case <-ctx.Done(): + // Clean shutdown: send close frame + c.connMu.Lock() + c.conn.WriteControl( + websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseNormalClosure, "shutdown"), + time.Now().Add(5*time.Second), + ) + c.connMu.Unlock() + return ctx.Err() + default: + } + + var frame wsFrame + if err := conn.ReadJSON(&frame); err != nil { + // Check if it's a close error + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Info("ws connection closed by server") + return nil + } + if websocket.IsUnexpectedCloseError(err) { + c.logger.Warn("ws connection unexpectedly closed", "error", err) + return err + } + return fmt.Errorf("read frame: %w", err) + } + + c.routeFrame(frame) + } +} + +// routeFrame dispatches a received frame to the appropriate handler. +func (c *WSClient) routeFrame(frame wsFrame) { + switch frame.Type { + case "res": + c.handleResponse(frame) + case "event": + c.handleEvent(frame) + default: + c.logger.Warn("unknown frame type", "type", frame.Type, "id", frame.ID) + } +} + +// handleResponse correlates a response frame to a pending request channel. +func (c *WSClient) handleResponse(frame wsFrame) { + c.mu.Lock() + ch, ok := c.pending[frame.ID] + if ok { + delete(c.pending, frame.ID) + } + c.mu.Unlock() + + if !ok { + c.logger.Warn("received response for unknown request", "id", frame.ID) + return + } + + if frame.Error != nil { + // Send nil to signal error; caller checks via Send return + ch <- nil + return + } + + ch <- frame.Result +} + +// handleEvent dispatches an event frame to registered handlers. +func (c *WSClient) handleEvent(frame wsFrame) { + c.mu.Lock() + handlers := c.handlers[frame.Event] + c.mu.Unlock() + + if len(handlers) == 0 { + c.logger.Debug("unhandled event", "event", frame.Event) + return + } + + for _, h := range handlers { + h(frame.Params) + } +} + +// ── Send ───────────────────────────────────────────────────────────────── + +// Send sends a JSON request to the gateway and returns the response payload. +// It is safe for concurrent use. The caller should check for errors in the +// returned payload. A nil payload with nil error means the gateway sent an +// error response (check via the response frame's error field, which is logged). +func (c *WSClient) Send(method string, params any) (json.RawMessage, error) { + reqID := uuid.New().String() + + paramsJSON, err := json.Marshal(params) + if err != nil { + return nil, fmt.Errorf("marshal params: %w", err) + } + + // Register pending response channel + respCh := make(chan json.RawMessage, 1) + c.mu.Lock() + c.pending[reqID] = respCh + c.mu.Unlock() + + defer func() { + c.mu.Lock() + delete(c.pending, reqID) + c.mu.Unlock() + }() + + // Build and send frame + frame := wsFrame{ + Type: "req", + ID: reqID, + Method: method, + Params: paramsJSON, + } + + c.connMu.Lock() + err = c.conn.WriteJSON(frame) + c.connMu.Unlock() + + if err != nil { + return nil, fmt.Errorf("write request: %w", err) + } + + // Wait for response with timeout + select { + case resp := <-respCh: + if resp == nil { + return nil, fmt.Errorf("gateway returned error for request %s", reqID) + } + return resp, nil + case <-time.After(30 * time.Second): + return nil, fmt.Errorf("request %s timed out", reqID) + } +} \ No newline at end of file