diff --git a/.claude/rules/bff-go.md b/.claude/rules/bff-go.md index e1e762be..6468e6ec 100644 --- a/.claude/rules/bff-go.md +++ b/.claude/rules/bff-go.md @@ -140,7 +140,7 @@ Env vars (some also available as CLI flags): | `AUTH_USER_HEADER` | `-auth-user-header` | `x-auth-request-user` | User header name | | `ADMIN_ROLE` | `-admin-role` | `admin` | OIDC role claim for admin (display gating only — gateway enforces) | | `LOGOUT_URL` | `-logout-url` | `/oauth2/sign_out` | Proxy sign-out path the frontend redirects to on logout | -| `FEATURE_*` | | varies | Feature flags: `FEATURE_TERMINAL`, `FEATURE_FILE_TRANSFER`, `FEATURE_SETTINGS`, `FEATURE_GLOBAL_POLICY`, `FEATURE_CREDENTIAL_REFRESH`, `FEATURE_SERVICES`, `FEATURE_DRAFT_POLICY` | +| `FEATURE_*` | | varies | Feature flags: `FEATURE_TERMINAL`, `FEATURE_FILE_TRANSFER`, `FEATURE_SETTINGS`, `FEATURE_GLOBAL_POLICY`, `FEATURE_CREDENTIAL_REFRESH`, `FEATURE_SERVICES`, `FEATURE_DRAFT_POLICY`, `FEATURE_LIVE_UPDATES` | ## Error handling diff --git a/.claude/rules/openshell-api.md b/.claude/rules/openshell-api.md index fc828d92..3e737d17 100644 --- a/.claude/rules/openshell-api.md +++ b/.claude/rules/openshell-api.md @@ -39,4 +39,4 @@ User-facing RPCs require `Bearer` (OIDC JWT). `Health` is unauthenticated. Sandb ## Streaming RPCs -`ExecSandboxInteractive` (bidi) — implemented via WebSocket relay in `terminal_handler.go`. `WatchSandbox` (server-stream) and `ForwardTcp` (bidi) — deferred; use `GetSandboxLogs` + `GetSandbox` polling instead. +`ExecSandboxInteractive` (bidi) — implemented via WebSocket relay in `terminal_handler.go`. `WatchSandbox` (server-stream) — implemented via WebSocket relay in `watch_handler.go`, gated by `FEATURE_LIVE_UPDATES` with React Query polling as fallback (ADR 0004). `ForwardTcp` (bidi) — deferred. diff --git a/FEATURE_FLAGS.md b/FEATURE_FLAGS.md index ec8e4e7e..0cee5fae 100644 --- a/FEATURE_FLAGS.md +++ b/FEATURE_FLAGS.md @@ -17,6 +17,7 @@ Flags are set as **BFF environment variables** and exposed to the frontend via ` | `FEATURE_CREDENTIAL_REFRESH` | `true` | Credential refresh configure/rotate/delete on provider detail. Disable if refresh is managed externally. | | `FEATURE_SERVICES` | `true` | Sandbox services (expose/list/delete). Disable if service routing is not available. | | `FEATURE_DRAFT_POLICY` | `true` | Draft policy advisor inbox (approve/reject/edit/undo). Disable if the policy advisor is not configured. | +| `FEATURE_LIVE_UPDATES` | `true` | Live sandbox/draft updates via a WatchSandbox WebSocket relay ([ADR 0004](docs/adrs/0004-live-updates-watch-relay.md)). When on, the sandbox detail page swaps polling for pushed events; polling resumes automatically if the socket cannot connect. Disable in federated deployments where the proxy can't upgrade WS. | ### Frontend consumption @@ -34,7 +35,8 @@ The `/api/v1/auth/config` response includes a `features` object: "globalPolicy": true, "credentialRefresh": true, "services": true, - "draftPolicy": true + "draftPolicy": true, + "liveUpdates": true } } ``` diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index f66d90bb..58f51f2b 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -78,6 +78,7 @@ func main() { CredentialRefresh: envOr("FEATURE_CREDENTIAL_REFRESH", "true") == "true", Services: envOr("FEATURE_SERVICES", "true") == "true", DraftPolicy: envOr("FEATURE_DRAFT_POLICY", "true") == "true", + LiveUpdates: envOr("FEATURE_LIVE_UPDATES", "true") == "true", }, } diff --git a/backend/internal/api/app.go b/backend/internal/api/app.go index 7552c574..e4608bba 100644 --- a/backend/internal/api/app.go +++ b/backend/internal/api/app.go @@ -89,6 +89,9 @@ func (app *App) Routes() http.Handler { r.Delete("/sandboxes/{name}", app.DeleteSandbox) r.Get("/sandboxes/{name}/logs", app.GetSandboxLogs) r.Get("/sandboxes/{name}/terminal", app.Terminal) + if app.authConfig.Features.LiveUpdates { + r.Get("/sandboxes/{name}/watch", app.WatchSandbox) + } r.Get("/sandboxes/{name}/providers", app.ListSandboxProviders) r.Post("/sandboxes/{name}/providers/{provider}", app.AttachSandboxProvider) r.Delete("/sandboxes/{name}/providers/{provider}", app.DetachSandboxProvider) diff --git a/backend/internal/api/gateway_handler.go b/backend/internal/api/gateway_handler.go index 07606339..e5ceb15e 100644 --- a/backend/internal/api/gateway_handler.go +++ b/backend/internal/api/gateway_handler.go @@ -31,6 +31,7 @@ type FeatureFlags struct { CredentialRefresh bool `json:"credentialRefresh"` Services bool `json:"services"` DraftPolicy bool `json:"draftPolicy"` + LiveUpdates bool `json:"liveUpdates"` } // AuthConfigResponse tells the frontend whether auth is enabled and which diff --git a/backend/internal/api/mock_gateway_test.go b/backend/internal/api/mock_gateway_test.go index 86cb5ec8..f256046d 100644 --- a/backend/internal/api/mock_gateway_test.go +++ b/backend/internal/api/mock_gateway_test.go @@ -16,6 +16,7 @@ type mockGateway struct { getGatewayInfoFn func(ctx context.Context) (*openshellv1.GetGatewayInfoResponse, error) getCurrentUserFn func(ctx context.Context) (*openshellv1.GetCurrentUserResponse, error) execSandboxInteractiveFn func(ctx context.Context) (openshellv1.OpenShell_ExecSandboxInteractiveClient, error) + watchSandboxFn func(ctx context.Context, req *openshellv1.WatchSandboxRequest) (openshellv1.OpenShell_WatchSandboxClient, error) createSandboxFn func(ctx context.Context, workspace, name string, spec *openshellv1.SandboxSpec, labels, annotations map[string]string) (*openshellv1.Sandbox, error) getSandboxFn func(ctx context.Context, workspace, name string) (*openshellv1.Sandbox, error) listSandboxesFn func(ctx context.Context, workspace string, limit, offset uint32, labelSelector string) ([]*openshellv1.Sandbox, error) @@ -83,6 +84,9 @@ func (m *mockGateway) GetCurrentUser(ctx context.Context) (*openshellv1.GetCurre func (m *mockGateway) ExecSandboxInteractive(ctx context.Context) (openshellv1.OpenShell_ExecSandboxInteractiveClient, error) { return m.execSandboxInteractiveFn(ctx) } +func (m *mockGateway) WatchSandbox(ctx context.Context, req *openshellv1.WatchSandboxRequest) (openshellv1.OpenShell_WatchSandboxClient, error) { + return m.watchSandboxFn(ctx, req) +} func (m *mockGateway) CreateSandbox(ctx context.Context, workspace, name string, spec *openshellv1.SandboxSpec, labels, annotations map[string]string) (*openshellv1.Sandbox, error) { return m.createSandboxFn(ctx, workspace, name, spec, labels, annotations) } diff --git a/backend/internal/api/watch_e2e_test.go b/backend/internal/api/watch_e2e_test.go new file mode 100644 index 00000000..465a53e9 --- /dev/null +++ b/backend/internal/api/watch_e2e_test.go @@ -0,0 +1,225 @@ +package api + +// End-to-end test of the live-updates path: a real gRPC server implementing +// the OpenShell service, a real gateway.Client dialing it over TCP, the real +// chi router (auth disabled, FEATURE_LIVE_UPDATES on), and a real WebSocket +// client. Verifies the whole chain — gRPC stream → gateway wrapper → WS +// relay → JSON frames — including log following. + +import ( + "context" + "encoding/json" + "net" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "google.golang.org/grpc" + + datamodelv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/datamodelv1" + openshellv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/openshellv1" + "github.com/Gkrumbach07/openshell-dashboard/backend/internal/auth" + "github.com/Gkrumbach07/openshell-dashboard/backend/internal/gateway" + "github.com/Gkrumbach07/openshell-dashboard/backend/internal/models" +) + +type e2eOpenShellServer struct { + openshellv1.UnimplementedOpenShellServer + // watchRequests records the WatchSandboxRequest each stream opened with. + watchRequests chan *openshellv1.WatchSandboxRequest + // events is drained into every open watch stream. + events chan *openshellv1.SandboxStreamEvent +} + +func e2eSandbox(phase openshellv1.SandboxPhase) *openshellv1.Sandbox { + return &openshellv1.Sandbox{ + Metadata: &datamodelv1.ObjectMeta{Id: "e2e-id-1", Name: "e2e-sandbox", Workspace: "default"}, + Status: &openshellv1.SandboxStatus{Phase: phase}, + } +} + +func (s *e2eOpenShellServer) GetSandbox(_ context.Context, _ *openshellv1.GetSandboxRequest) (*openshellv1.SandboxResponse, error) { + return &openshellv1.SandboxResponse{Sandbox: e2eSandbox(openshellv1.SandboxPhase_SANDBOX_PHASE_READY)}, nil +} + +func (s *e2eOpenShellServer) WatchSandbox(req *openshellv1.WatchSandboxRequest, stream grpc.ServerStreamingServer[openshellv1.SandboxStreamEvent]) error { + s.watchRequests <- req + // Initial snapshot, mirroring the real gateway's producer. + if err := stream.Send(&openshellv1.SandboxStreamEvent{ + Payload: &openshellv1.SandboxStreamEvent_Sandbox{ + Sandbox: e2eSandbox(openshellv1.SandboxPhase_SANDBOX_PHASE_READY), + }, + }); err != nil { + return err + } + for { + select { + case evt := <-s.events: + if err := stream.Send(evt); err != nil { + return err + } + case <-stream.Context().Done(): + return stream.Context().Err() + } + } +} + +// startE2EStack boots the fake gRPC gateway, a real gateway.Client, and the +// real router, returning the WS URL base and the fake server for triggering +// events. +func startE2EStack(t *testing.T) (string, *e2eOpenShellServer) { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + fake := &e2eOpenShellServer{ + watchRequests: make(chan *openshellv1.WatchSandboxRequest, 4), + events: make(chan *openshellv1.SandboxStreamEvent, 16), + } + grpcServer := grpc.NewServer() + openshellv1.RegisterOpenShellServer(grpcServer, fake) + go func() { _ = grpcServer.Serve(lis) }() + t.Cleanup(grpcServer.Stop) + + gatewayClient, err := gateway.New(lis.Addr().String(), "") + if err != nil { + t.Fatalf("gateway client: %v", err) + } + t.Cleanup(func() { _ = gatewayClient.Close() }) + + app := NewApp( + gatewayClient, + auth.New(auth.Config{Disabled: true}), + "", + AuthConfigResponse{AuthDisabled: true, Features: FeatureFlags{LiveUpdates: true}}, + ) + server := httptest.NewServer(app.Routes()) + t.Cleanup(server.Close) + + return "ws" + strings.TrimPrefix(server.URL, "http"), fake +} + +func readWatchEvent(t *testing.T, ws *websocket.Conn) models.WatchEvent { + t.Helper() + _ = ws.SetReadDeadline(time.Now().Add(5 * time.Second)) + _, data, err := ws.ReadMessage() + if err != nil { + t.Fatalf("read frame: %v", err) + } + var evt models.WatchEvent + if err := json.Unmarshal(data, &evt); err != nil { + t.Fatalf("unmarshal frame %q: %v", data, err) + } + return evt +} + +// requireWatchRequest asserts the relay resolved name → id and opened the +// stream with the right follow options. +func requireWatchRequest(t *testing.T, fake *e2eOpenShellServer) { + t.Helper() + select { + case req := <-fake.watchRequests: + if req.GetId() != "e2e-id-1" || !req.GetFollowStatus() || !req.GetFollowLogs() || + req.GetLogTailLines() != 50 || req.GetLogMinLevel() != "INFO" { + t.Errorf("watch request = %+v, want id=e2e-id-1 followStatus followLogs lines=50 level=INFO", req) + } + case <-time.After(5 * time.Second): + t.Fatal("gateway never received a WatchSandbox request") + } +} + +func TestE2EWatchRelaysSnapshotsAndLogs(t *testing.T) { + wsBase, fake := startE2EStack(t) + + ws, _, err := websocket.DefaultDialer.Dial( + wsBase+"/api/v1/workspaces/default/sandboxes/e2e-sandbox/watch?logs=true&lines=50&level=INFO", + nil, + ) + if err != nil { + t.Fatalf("ws dial: %v", err) + } + defer ws.Close() + + requireWatchRequest(t, fake) + + // Initial snapshot arrives through the whole chain. + first := readWatchEvent(t, ws) + if first.Type != "sandbox" || first.Sandbox == nil || first.Sandbox.Metadata.ID != "e2e-id-1" { + t.Fatalf("first frame = %+v, want initial sandbox snapshot", first) + } + + // A pushed status change (what the gateway emits on a draft mutation). + fake.events <- &openshellv1.SandboxStreamEvent{ + Payload: &openshellv1.SandboxStreamEvent_Sandbox{ + Sandbox: e2eSandbox(openshellv1.SandboxPhase_SANDBOX_PHASE_READY), + }, + } + second := readWatchEvent(t, ws) + if second.Type != "sandbox" { + t.Fatalf("second frame type = %q, want sandbox", second.Type) + } + + // A pushed log line (Phase 3 follow_logs path). + fake.events <- &openshellv1.SandboxStreamEvent{ + Payload: &openshellv1.SandboxStreamEvent_Log{ + Log: &openshellv1.SandboxLogLine{ + SandboxId: "e2e-id-1", + TimestampMs: 42, + Level: "INFO", + Message: "network decision", + Source: "sandbox", + Fields: map[string]string{"dst_host": "google.com", "action": "deny"}, + }, + }, + } + third := readWatchEvent(t, ws) + if third.Type != "log" || third.Log == nil || third.Log.Message != "network decision" || + third.Log.Fields["dst_host"] != "google.com" { + t.Fatalf("third frame = %+v, want relayed log line", third) + } +} + +func TestE2EWatchRouteAbsentWhenFlagOff(t *testing.T) { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + openshellv1.RegisterOpenShellServer(grpcServer, &e2eOpenShellServer{ + watchRequests: make(chan *openshellv1.WatchSandboxRequest, 1), + events: make(chan *openshellv1.SandboxStreamEvent, 1), + }) + go func() { _ = grpcServer.Serve(lis) }() + t.Cleanup(grpcServer.Stop) + + gatewayClient, err := gateway.New(lis.Addr().String(), "") + if err != nil { + t.Fatalf("gateway client: %v", err) + } + t.Cleanup(func() { _ = gatewayClient.Close() }) + + app := NewApp( + gatewayClient, + auth.New(auth.Config{Disabled: true}), + "", + AuthConfigResponse{AuthDisabled: true, Features: FeatureFlags{LiveUpdates: false}}, + ) + server := httptest.NewServer(app.Routes()) + t.Cleanup(server.Close) + + wsBase := "ws" + strings.TrimPrefix(server.URL, "http") + _, resp, err := websocket.DefaultDialer.Dial( + wsBase+"/api/v1/workspaces/default/sandboxes/e2e-sandbox/watch", + nil, + ) + if err == nil { + t.Fatal("dial succeeded, want the watch route to be absent when the flag is off") + } + if resp == nil || resp.StatusCode != 404 { + t.Fatalf("dial response = %+v, want 404", resp) + } +} diff --git a/backend/internal/api/watch_handler.go b/backend/internal/api/watch_handler.go new file mode 100644 index 00000000..1957a3d6 --- /dev/null +++ b/backend/internal/api/watch_handler.go @@ -0,0 +1,174 @@ +package api + +import ( + "context" + "log/slog" + "net/http" + "net/url" + "strconv" + "time" + + "github.com/go-chi/chi/v5" + "github.com/gorilla/websocket" + + openshellv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/openshellv1" + "github.com/Gkrumbach07/openshell-dashboard/backend/internal/models" +) + +const ( + // watchWriteWait bounds a single WebSocket write. + watchWriteWait = 10 * time.Second + // watchPongWait is how long the client has to answer a ping. Unlike the + // terminal relay, a watch stream can be silent for minutes, so keepalive + // pings are the only way to detect a half-dead proxy connection. + watchPongWait = 60 * time.Second + // watchPingPeriod must be shorter than watchPongWait. + watchPingPeriod = 50 * time.Second + // watchEventBuffer absorbs bursts from the gateway watch bus while a + // slow client drains earlier frames. + watchEventBuffer = 32 +) + +// buildWatchRequest translates query params into the gRPC watch request. +// Status snapshots are always followed; log following is opt-in with the +// same filters as GET .../logs — lines (tail replay, default 200), sinceMs, +// source (repeatable: gateway|sandbox), level (min level, e.g. INFO). +func buildWatchRequest(sandboxID string, query url.Values) *openshellv1.WatchSandboxRequest { + req := &openshellv1.WatchSandboxRequest{ + Id: sandboxID, + FollowStatus: true, + } + if query.Get("logs") != "true" { + return req + } + req.FollowLogs = true + req.LogTailLines = 200 + if raw := query.Get("lines"); raw != "" { + if parsed, err := strconv.ParseUint(raw, 10, 32); err == nil { + req.LogTailLines = uint32(parsed) + } + } + if raw := query.Get("sinceMs"); raw != "" { + if parsed, err := strconv.ParseInt(raw, 10, 64); err == nil { + req.LogSinceMs = parsed + } + } + req.LogSources = query["source"] + req.LogMinLevel = query.Get("level") + return req +} + +// startWatchReader pumps the client side of the socket: the client sends no +// application messages, but reading is required to process pong control +// frames and detect close. Cancels ctx when the client goes away. +func startWatchReader(ws *websocket.Conn, cancel context.CancelFunc) { + go func() { + ws.SetReadLimit(512) + _ = ws.SetReadDeadline(time.Now().Add(watchPongWait)) + ws.SetPongHandler(func(string) error { + return ws.SetReadDeadline(time.Now().Add(watchPongWait)) + }) + for { + if _, _, err := ws.ReadMessage(); err != nil { + cancel() + return + } + } + }() +} + +// pumpWatchEvents drains the gRPC stream into a channel of wire DTOs, +// closing it when the stream ends. +func pumpWatchEvents(ctx context.Context, stream openshellv1.OpenShell_WatchSandboxClient, events chan<- models.WatchEvent) { + defer close(events) + for { + evt, err := stream.Recv() + if err != nil { + return + } + frame, ok := models.FromSandboxStreamEvent(evt) + if !ok { + continue + } + select { + case events <- frame: + case <-ctx.Done(): + return + } + } +} + +// writeWatchFrames owns all writes to the socket (gorilla/websocket allows +// one concurrent writer): JSON data frames plus keepalive pings. +func writeWatchFrames(ws *websocket.Conn, events <-chan models.WatchEvent) { + ticker := time.NewTicker(watchPingPeriod) + defer ticker.Stop() + for { + select { + case frame, ok := <-events: + if !ok { + _ = ws.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseNormalClosure, "stream ended"), + time.Now().Add(watchWriteWait)) + return + } + _ = ws.SetWriteDeadline(time.Now().Add(watchWriteWait)) + if err := ws.WriteJSON(frame); err != nil { + return + } + case <-ticker.C: + _ = ws.SetWriteDeadline(time.Now().Add(watchWriteWait)) + if err := ws.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} + +// WatchSandbox relays the gateway's server-streaming WatchSandbox RPC over a +// WebSocket as JSON WatchEvent frames. See buildWatchRequest for the query +// params. WatchSandbox (gRPC) takes sandbox_id, not name — the BFF resolves +// name → metadata.id via GetSandbox. +func (app *App) WatchSandbox(w http.ResponseWriter, r *http.Request) { + workspace := chi.URLParam(r, "workspace") + name := chi.URLParam(r, "name") + + sandbox, err := app.gateway.GetSandbox(r.Context(), workspace, name) + if err != nil { + writeGrpcError(w, err) + return + } + sandboxID := sandbox.GetMetadata().GetId() + if sandboxID == "" { + writeError(w, http.StatusNotFound, "not_found", "sandbox has no id") + return + } + req := buildWatchRequest(sandboxID, r.URL.Query()) + + upgrader := websocket.Upgrader{ + CheckOrigin: checkWebSocketOrigin, + } + ws, err := upgrader.Upgrade(w, r, nil) + if err != nil { + slog.Error("watch websocket upgrade failed", "error", err) + return + } + defer ws.Close() + + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + + stream, err := app.gateway.WatchSandbox(ctx, req) + if err != nil { + slog.Error("watch stream open failed", "sandbox_id", sandboxID, "error", err) + _ = ws.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "failed to open watch stream"), + time.Now().Add(watchWriteWait)) + return + } + + startWatchReader(ws, cancel) + events := make(chan models.WatchEvent, watchEventBuffer) + go pumpWatchEvents(ctx, stream, events) + writeWatchFrames(ws, events) +} diff --git a/backend/internal/api/watch_handler_test.go b/backend/internal/api/watch_handler_test.go new file mode 100644 index 00000000..8ba3287c --- /dev/null +++ b/backend/internal/api/watch_handler_test.go @@ -0,0 +1,186 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/go-chi/chi/v5" + "github.com/gorilla/websocket" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + datamodelv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/datamodelv1" + openshellv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/openshellv1" + "github.com/Gkrumbach07/openshell-dashboard/backend/internal/models" +) + +// fakeWatchStream implements openshellv1.OpenShell_WatchSandboxClient, +// serving a fixed list of events then blocking until the context ends. +type fakeWatchStream struct { + grpc.ClientStream + ctx context.Context + events []*openshellv1.SandboxStreamEvent + index int +} + +func (s *fakeWatchStream) Recv() (*openshellv1.SandboxStreamEvent, error) { + if s.index < len(s.events) { + evt := s.events[s.index] + s.index++ + return evt, nil + } + <-s.ctx.Done() + return nil, s.ctx.Err() +} + +func (s *fakeWatchStream) Header() (metadata.MD, error) { return nil, nil } +func (s *fakeWatchStream) Trailer() metadata.MD { return nil } +func (s *fakeWatchStream) CloseSend() error { return nil } +func (s *fakeWatchStream) Context() context.Context { return s.ctx } + +func watchTestServer(t *testing.T, gw *mockGateway) *httptest.Server { + t.Helper() + app := newTestApp(gw) + r := chi.NewRouter() + r.Get("/workspaces/{workspace}/sandboxes/{name}/watch", app.WatchSandbox) + server := httptest.NewServer(r) + t.Cleanup(server.Close) + return server +} + +// watchRequestOrFail returns the request the relay opened its stream with, +// delivered over a channel because the mock runs on the server goroutine. +func watchRequestOrFail(t *testing.T, reqCh <-chan *openshellv1.WatchSandboxRequest) *openshellv1.WatchSandboxRequest { + t.Helper() + select { + case req := <-reqCh: + return req + case <-time.After(5 * time.Second): + t.Fatal("watch stream was never opened") + return nil + } +} + +func TestWatchSandboxRelaysEvents(t *testing.T) { + reqCh := make(chan *openshellv1.WatchSandboxRequest, 1) + gw := &mockGateway{ + getSandboxFn: func(_ context.Context, _, _ string) (*openshellv1.Sandbox, error) { + return &openshellv1.Sandbox{ + Metadata: &datamodelv1.ObjectMeta{Id: "sb-id-1", Name: "my-sandbox", Workspace: "default"}, + }, nil + }, + watchSandboxFn: func(ctx context.Context, req *openshellv1.WatchSandboxRequest) (openshellv1.OpenShell_WatchSandboxClient, error) { + reqCh <- req + return &fakeWatchStream{ + ctx: ctx, + events: []*openshellv1.SandboxStreamEvent{ + {Payload: &openshellv1.SandboxStreamEvent_Sandbox{ + Sandbox: &openshellv1.Sandbox{ + Metadata: &datamodelv1.ObjectMeta{Id: "sb-id-1", Name: "my-sandbox", Workspace: "default"}, + Status: &openshellv1.SandboxStatus{Phase: openshellv1.SandboxPhase_SANDBOX_PHASE_READY}, + }, + }}, + {Payload: &openshellv1.SandboxStreamEvent_Warning{ + Warning: &openshellv1.SandboxStreamWarning{Message: "lagged"}, + }}, + }, + }, nil + }, + } + server := watchTestServer(t, gw) + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/workspaces/default/sandboxes/my-sandbox/watch" + ws, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial failed: %v", err) + } + defer ws.Close() + + _ = ws.SetReadDeadline(time.Now().Add(5 * time.Second)) + + var first models.WatchEvent + if _, data, readErr := ws.ReadMessage(); readErr != nil { + t.Fatalf("read first frame: %v", readErr) + } else if unmarshalErr := json.Unmarshal(data, &first); unmarshalErr != nil { + t.Fatalf("unmarshal first frame: %v", unmarshalErr) + } + if first.Type != "sandbox" || first.Sandbox == nil || first.Sandbox.Status.Phase != "READY" { + t.Errorf("first frame = %+v, want sandbox snapshot with READY phase", first) + } + + var second models.WatchEvent + if _, data, readErr := ws.ReadMessage(); readErr != nil { + t.Fatalf("read second frame: %v", readErr) + } else if unmarshalErr := json.Unmarshal(data, &second); unmarshalErr != nil { + t.Fatalf("unmarshal second frame: %v", unmarshalErr) + } + if second.Type != "warning" || second.Warning != "lagged" { + t.Errorf("second frame = %+v, want warning 'lagged'", second) + } + + gotReq := watchRequestOrFail(t, reqCh) + if gotReq.GetId() != "sb-id-1" { + t.Errorf("watch request id = %q, want sb-id-1 (resolved from name)", gotReq.GetId()) + } + if !gotReq.GetFollowStatus() { + t.Error("watch request should always follow status") + } + if gotReq.GetFollowLogs() { + t.Error("watch request should not follow logs unless requested") + } +} + +func TestWatchSandboxLogParams(t *testing.T) { + reqCh := make(chan *openshellv1.WatchSandboxRequest, 1) + gw := &mockGateway{ + getSandboxFn: func(_ context.Context, _, _ string) (*openshellv1.Sandbox, error) { + return &openshellv1.Sandbox{ + Metadata: &datamodelv1.ObjectMeta{Id: "sb-id-1", Name: "my-sandbox", Workspace: "default"}, + }, nil + }, + watchSandboxFn: func(ctx context.Context, req *openshellv1.WatchSandboxRequest) (openshellv1.OpenShell_WatchSandboxClient, error) { + reqCh <- req + return &fakeWatchStream{ctx: ctx}, nil + }, + } + server := watchTestServer(t, gw) + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + "/workspaces/default/sandboxes/my-sandbox/watch?logs=true&lines=50&level=WARN&source=sandbox" + ws, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial failed: %v", err) + } + defer ws.Close() + + gotReq := watchRequestOrFail(t, reqCh) + if !gotReq.GetFollowLogs() || gotReq.GetLogTailLines() != 50 || + gotReq.GetLogMinLevel() != "WARN" || len(gotReq.GetLogSources()) != 1 || gotReq.GetLogSources()[0] != "sandbox" { + t.Errorf("watch request = %+v, want followLogs with lines=50 level=WARN source=sandbox", gotReq) + } +} + +func TestWatchSandboxGetSandboxError(t *testing.T) { + gw := &mockGateway{ + getSandboxFn: func(_ context.Context, _, _ string) (*openshellv1.Sandbox, error) { + return nil, status.Error(codes.NotFound, "no such sandbox") + }, + } + server := watchTestServer(t, gw) + + resp, err := http.Get(server.URL + "/workspaces/default/sandboxes/missing/watch") + if err != nil { + t.Fatalf("request failed: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusNotFound) + } +} diff --git a/backend/internal/gateway/interface.go b/backend/internal/gateway/interface.go index 1a468f4f..f0b7fc7d 100644 --- a/backend/internal/gateway/interface.go +++ b/backend/internal/gateway/interface.go @@ -16,6 +16,7 @@ type Interface interface { GetGatewayInfo(ctx context.Context) (*openshellv1.GetGatewayInfoResponse, error) GetCurrentUser(ctx context.Context) (*openshellv1.GetCurrentUserResponse, error) ExecSandboxInteractive(ctx context.Context) (openshellv1.OpenShell_ExecSandboxInteractiveClient, error) + WatchSandbox(ctx context.Context, req *openshellv1.WatchSandboxRequest) (openshellv1.OpenShell_WatchSandboxClient, error) // Sandboxes CreateSandbox(ctx context.Context, workspace, name string, spec *openshellv1.SandboxSpec, labels, annotations map[string]string) (*openshellv1.Sandbox, error) diff --git a/backend/internal/gateway/watch.go b/backend/internal/gateway/watch.go new file mode 100644 index 00000000..9aae529e --- /dev/null +++ b/backend/internal/gateway/watch.go @@ -0,0 +1,13 @@ +package gateway + +import ( + "context" + + openshellv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/openshellv1" +) + +// WatchSandbox opens the server-streaming WatchSandbox RPC. The caller owns +// the stream lifecycle: cancel ctx to stop the gateway-side producer. +func (c *Client) WatchSandbox(ctx context.Context, req *openshellv1.WatchSandboxRequest) (openshellv1.OpenShell_WatchSandboxClient, error) { + return c.openshell.WatchSandbox(ctx, req) +} diff --git a/backend/internal/models/observability.go b/backend/internal/models/observability.go index 2eaa7cb4..f8cbf68a 100644 --- a/backend/internal/models/observability.go +++ b/backend/internal/models/observability.go @@ -29,18 +29,22 @@ type SandboxLogs struct { BufferTotal uint32 `json:"bufferTotal"` } +func FromLogLine(line *openshellv1.SandboxLogLine) LogLine { + return LogLine{ + SandboxID: line.GetSandboxId(), + TimestampMs: line.GetTimestampMs(), + Level: line.GetLevel(), + Target: line.GetTarget(), + Message: line.GetMessage(), + Source: line.GetSource(), + Fields: line.GetFields(), + } +} + func FromSandboxLogs(resp *openshellv1.GetSandboxLogsResponse) SandboxLogs { out := SandboxLogs{Logs: []LogLine{}, BufferTotal: resp.GetBufferTotal()} for _, line := range resp.GetLogs() { - out.Logs = append(out.Logs, LogLine{ - SandboxID: line.SandboxId, - TimestampMs: line.TimestampMs, - Level: line.Level, - Target: line.Target, - Message: line.Message, - Source: line.Source, - Fields: line.Fields, - }) + out.Logs = append(out.Logs, FromLogLine(line)) } return out } diff --git a/backend/internal/models/watch.go b/backend/internal/models/watch.go new file mode 100644 index 00000000..e8a8d139 --- /dev/null +++ b/backend/internal/models/watch.go @@ -0,0 +1,48 @@ +package models + +import ( + openshellv1 "github.com/Gkrumbach07/openshell-dashboard/backend/gen/openshellv1" +) + +// WatchEvent is one JSON text frame on the sandbox watch WebSocket. Type is +// the discriminator: "sandbox", "log", "warning", or "draftPolicyUpdate"; +// exactly one payload field is set. Payloads go through the From*() +// converters so secret-field stripping applies — never serialize raw proto. +type WatchEvent struct { + Sandbox *Sandbox `json:"sandbox,omitempty"` + Log *LogLine `json:"log,omitempty"` + DraftPolicyUpdate *DraftPolicyUpdate `json:"draftPolicyUpdate,omitempty"` + Type string `json:"type"` + Warning string `json:"warning,omitempty"` +} + +// DraftPolicyUpdate mirrors openshell.v1.DraftPolicyUpdate. The gateway does +// not emit this payload yet; it is relayed for forward compatibility. +type DraftPolicyUpdate struct { + DraftVersion uint64 `json:"draftVersion"` + NewChunks uint32 `json:"newChunks"` + TotalPending uint32 `json:"totalPending"` +} + +// FromSandboxStreamEvent converts one gateway stream event into the wire DTO. +// ok is false for payloads the dashboard does not relay (platform events) — +// callers skip those frames. +func FromSandboxStreamEvent(evt *openshellv1.SandboxStreamEvent) (WatchEvent, bool) { + switch p := evt.GetPayload().(type) { + case *openshellv1.SandboxStreamEvent_Sandbox: + sandbox := FromSandbox(p.Sandbox) + return WatchEvent{Type: "sandbox", Sandbox: &sandbox}, true + case *openshellv1.SandboxStreamEvent_Log: + log := FromLogLine(p.Log) + return WatchEvent{Type: "log", Log: &log}, true + case *openshellv1.SandboxStreamEvent_Warning: + return WatchEvent{Type: "warning", Warning: p.Warning.GetMessage()}, true + case *openshellv1.SandboxStreamEvent_DraftPolicyUpdate: + return WatchEvent{Type: "draftPolicyUpdate", DraftPolicyUpdate: &DraftPolicyUpdate{ + DraftVersion: p.DraftPolicyUpdate.GetDraftVersion(), + NewChunks: p.DraftPolicyUpdate.GetNewChunks(), + TotalPending: p.DraftPolicyUpdate.GetTotalPending(), + }}, true + } + return WatchEvent{}, false +} diff --git a/docs/adrs/0004-live-updates-watch-relay.md b/docs/adrs/0004-live-updates-watch-relay.md new file mode 100644 index 00000000..552c4495 --- /dev/null +++ b/docs/adrs/0004-live-updates-watch-relay.md @@ -0,0 +1,69 @@ +# ADR 0004: Live Updates — WatchSandbox WebSocket Relay with Polling Fallback + +**Status:** Accepted +**Date:** 2026-08-10 +**Authors:** Gage Krumbach + +## Context + +The dashboard originally learned about sandbox and draft-policy changes only +by polling: sandbox detail every 5s, the draft inbox every 10s. Gateway-side +actions — a draft chunk landing from denial analysis, an auto-approval +(`proposal_approval_mode=auto`), a policy revision loading — took up to a +poll interval to appear, which users read as "I have to refresh." + +The gateway already has a real-time primitive: `WatchSandbox`, a per-sandbox +server-streaming RPC (bearer auth, `sandbox:read`). Its watch bus fires on +every draft mutation and policy revision persist, and the stream can also +carry filtered log lines and platform events. The dashboard already relays +one streaming RPC over WebSocket (`ExecSandboxInteractive`, the terminal). + +WebSockets do not work in every deployment: federated proxies that cannot +upgrade WS are exactly why `FEATURE_TERMINAL` exists. + +## Decision + +**Relay `WatchSandbox` over a WebSocket per open sandbox detail page, gated +by `FEATURE_LIVE_UPDATES`, with React Query polling as the automatic +fallback.** + +- The BFF exposes `GET .../sandboxes/{name}/watch` (`watch_handler.go`), + registered only when the flag is on. It resolves name → id, opens + `WatchSandbox` with `follow_status` always on (logs opt-in via query + params), and relays events as JSON `WatchEvent` frames converted through + `models.From*()` so secret-stripping applies. Ping/pong keepalive detects + half-dead proxy connections — unlike the terminal, a watch stream can be + legitimately silent for minutes. +- The frontend `useSandboxWatch` hook owns the detail page's socket. React + Query remains the single source of truth: snapshot events are pushed into + the sandbox detail cache, everything else becomes debounced query + invalidation. The logs tab opens its own socket with `logs=true` + (`useSandboxLogStream`) so log streaming only runs while that tab is + mounted; both hooks share one reconnecting-socket helper (`watchSocket.ts`). +- Queries accept `{ live }` and disable `refetchInterval` while the socket + is open. Flag off, socket down, or repeated connect failures → intervals + resume. There is no third mode: WebSocket-when-possible, polling otherwise. +- The server does not yet emit the proto's `DraftPolicyUpdate` payload; the + relay forwards it for forward-compatibility but the hook keys off + `sandbox` snapshots and `warning` frames. + +## Alternatives considered + +- **Shorter poll intervals** — multiplies request load for latency still + bounded by the interval; rejected. +- **SSE from the BFF** — avoids WS-upgrade problems, but adds a second + streaming transport next to the existing terminal WS for little gain. +- **One workspace-wide socket** — no upstream primitive; `WatchSandbox` is + per-sandbox. List pages and the draft summary keep polling. + +## Consequences + +- Draft/policy changes appear sub-second after the gateway processes them + (the in-sandbox denial aggregator's batching delay remains — that is + upstream behavior). +- Latency is the win, not request volume: events are "changed" pings that + drive refetches. +- One long-lived goroutine pair and gRPC stream per open detail page; the + stream is torn down when the page unmounts or the socket dies. +- Deployments behind non-WS proxies set `FEATURE_LIVE_UPDATES=false` (or + just let the hook give up) and silently keep today's polling behavior. diff --git a/docs/adrs/README.md b/docs/adrs/README.md index 73274684..d5a6f0cf 100644 --- a/docs/adrs/README.md +++ b/docs/adrs/README.md @@ -9,5 +9,6 @@ recorded here — an ADR appears when the decision does. | [0001](0001-downstream-consumption.md) | Downstream consumption: npm package + a five-mechanism extension surface | Accepted | | [0002](0002-auth-relay-only-bff.md) | Auth: relay-only BFF behind a fronting proxy (e.g. oauth2-proxy) | Accepted | | [0003](0003-gateway-client-sdk-vs-stubs.md) | Gateway client: openshell-sdk-go over generated stubs | **Proposed** | +| [0004](0004-live-updates-watch-relay.md) | Live updates: WatchSandbox WebSocket relay with polling fallback | Accepted | From here forward, ADRs are append-only: supersede, don't rewrite. diff --git a/frontend/src/api/auth.ts b/frontend/src/api/auth.ts index b243eb19..33f1f89a 100644 --- a/frontend/src/api/auth.ts +++ b/frontend/src/api/auth.ts @@ -37,6 +37,7 @@ export const useFeatureFlags = () => { credentialRefresh: true, services: true, draftPolicy: true, + liveUpdates: true, }; return data?.features ?? defaults; }; diff --git a/frontend/src/api/policy.ts b/frontend/src/api/policy.ts index 0f9bd2db..aa960eb3 100644 --- a/frontend/src/api/policy.ts +++ b/frontend/src/api/policy.ts @@ -161,11 +161,17 @@ export const useDeleteGlobalPolicy = () => { }); }; -export const useDraftPolicy = (workspace: string, name: string) => +// When `live` is true, a watch WebSocket (useSandboxWatch) is pushing +// invalidations, so interval polling is disabled and acts only as fallback. +export const useDraftPolicy = ( + workspace: string, + name: string, + opts?: { live?: boolean }, +) => useQuery({ queryKey: policyKeys.drafts(workspace, name), queryFn: () => getDraftPolicy(workspace, name), - refetchInterval: DRAFT_POLL_MS, + refetchInterval: opts?.live ? false : DRAFT_POLL_MS, }); // Draft decisions invalidate both the inbox and the policy view (approvals diff --git a/frontend/src/api/sandboxes.ts b/frontend/src/api/sandboxes.ts index c3ab6fb3..f3545d28 100644 --- a/frontend/src/api/sandboxes.ts +++ b/frontend/src/api/sandboxes.ts @@ -52,11 +52,17 @@ export const useSandboxes = (workspace: string, labelSelector?: string) => refetchInterval: SANDBOX_POLL_MS, }); -export const useSandbox = (workspace: string, name: string) => +// When `live` is true, a watch WebSocket (useSandboxWatch) is pushing +// sandbox snapshots into this cache, so interval polling is disabled. +export const useSandbox = ( + workspace: string, + name: string, + opts?: { live?: boolean }, +) => useQuery({ queryKey: sandboxKeys.detail(workspace, name), queryFn: () => getSandbox(workspace, name), - refetchInterval: SANDBOX_POLL_MS, + refetchInterval: opts?.live ? false : SANDBOX_POLL_MS, }); export const useCreateSandbox = (workspace: string) => { diff --git a/frontend/src/components/SandboxDraftsTab/index.tsx b/frontend/src/components/SandboxDraftsTab/index.tsx index 3191c5f8..a501cef3 100644 --- a/frontend/src/components/SandboxDraftsTab/index.tsx +++ b/frontend/src/components/SandboxDraftsTab/index.tsx @@ -26,14 +26,16 @@ import type { PolicyChunk } from '../../types'; type SandboxDraftsTabProps = { workspace: string; sandboxName: string; + isLive?: boolean; }; const SandboxDraftsTab: React.FC = ({ workspace, sandboxName, + isLive = false, }) => { const { isWorkspaceAdmin } = useWorkspaceRole(workspace); - const drafts = useDraftPolicy(workspace, sandboxName); + const drafts = useDraftPolicy(workspace, sandboxName, { live: isLive }); const actions = useDraftActions(workspace, sandboxName); const [rejecting, setRejecting] = useState(null); const [rejectReason, setRejectReason] = useState(''); diff --git a/frontend/src/components/sandbox/SandboxLogsTab.tsx b/frontend/src/components/sandbox/SandboxLogsTab.tsx index 0e9fa1ec..4b67060d 100644 --- a/frontend/src/components/sandbox/SandboxLogsTab.tsx +++ b/frontend/src/components/sandbox/SandboxLogsTab.tsx @@ -6,6 +6,7 @@ import { Checkbox, FormSelect, FormSelectOption, + Label, Toolbar, ToolbarContent, ToolbarGroup, @@ -14,6 +15,7 @@ import { import { LogViewer, LogViewerSearch } from '@patternfly/react-log-viewer'; import { useSandboxLogs } from '../../api/sandboxes'; +import { useSandboxLogStream } from '../../hooks/useSandboxLogStream'; import type { LogLine } from '../../types'; type SandboxLogsTabProps = { @@ -39,6 +41,18 @@ const SandboxLogsTab: React.FC = ({ const [source, setSource] = useState(''); const [lines, setLines] = useState(DEFAULT_LOG_LINES); const [autoRefresh, setAutoRefresh] = useState(true); + // Streamed lines over the watch WebSocket replace polling while the + // socket is healthy (ADR 0004); the polled query below is the fallback. + const stream = useSandboxLogStream( + workspace, + sandboxName, + { + lines: Number(lines), + level: level || undefined, + sources: source ? [source] : undefined, + }, + true, + ); const logs = useSandboxLogs( workspace, sandboxName, @@ -47,14 +61,16 @@ const SandboxLogsTab: React.FC = ({ level: level || undefined, sources: source ? [source] : undefined, }, - autoRefresh, + autoRefresh && !stream.isStreaming, ); const logText = useMemo(() => { - const logLines = logs.data?.logs ?? []; + const logLines = stream.isStreaming + ? stream.lines + : (logs.data?.logs ?? []); if (logLines.length === 0) return 'No log lines match the current filters.'; return logLines.map(formatLogLine).join('\n'); - }, [logs.data]); + }, [stream.isStreaming, stream.lines, logs.data]); const toolbar = ( @@ -100,13 +116,19 @@ const SandboxLogsTab: React.FC = ({ - setAutoRefresh(checked)} - /> + {stream.isStreaming ? ( + + ) : ( + setAutoRefresh(checked)} + /> + )} {(level !== '' || source !== '' || lines !== '200') && ( @@ -133,7 +155,8 @@ const SandboxLogsTab: React.FC = ({ ); - if (logs.isError) { + // A failed poll only matters when the stream isn't delivering lines. + if (logs.isError && !stream.isStreaming) { return ( = ({ return ( ({ + useFeatureFlags: () => mockUseFeatureFlags(), +})); + +class FakeWebSocket { + static instances: FakeWebSocket[] = []; + url: string; + onopen: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + onclose: (() => void) | null = null; + onerror: (() => void) | null = null; + closed = false; + + constructor(url: string) { + this.url = url; + FakeWebSocket.instances.push(this); + } + + close() { + this.closed = true; + this.onclose?.(); + } +} + +const logFrame = (message: string) => + JSON.stringify({ + type: 'log', + log: { message, timestampMs: 1, level: 'INFO' }, + }); + +describe('useSandboxLogStream', () => { + const realWebSocket = window.WebSocket; + + beforeEach(() => { + jest.useFakeTimers(); + FakeWebSocket.instances = []; + (window as unknown as { WebSocket: unknown }).WebSocket = FakeWebSocket; + mockUseFeatureFlags.mockReturnValue({ liveUpdates: true }); + }); + + afterEach(() => { + jest.useRealTimers(); + (window as unknown as { WebSocket: unknown }).WebSocket = realWebSocket; + }); + + it('does not open a socket when disabled or flag is off', () => { + const { rerender } = renderHook( + ({ enabled }) => + useSandboxLogStream('default', 'sb1', { lines: 100 }, enabled), + { initialProps: { enabled: false } }, + ); + expect(FakeWebSocket.instances).toHaveLength(0); + + mockUseFeatureFlags.mockReturnValue({ liveUpdates: false }); + rerender({ enabled: true }); + expect(FakeWebSocket.instances).toHaveLength(0); + }); + + it('opens a socket with log params and collects streamed lines', () => { + const { result } = renderHook(() => + useSandboxLogStream( + 'default', + 'sb1', + { lines: 100, level: 'WARN', sources: ['sandbox'] }, + true, + ), + ); + const ws = FakeWebSocket.instances[0]; + expect(ws.url).toContain('/sandboxes/sb1/watch?'); + expect(ws.url).toContain('logs=true'); + expect(ws.url).toContain('lines=100'); + expect(ws.url).toContain('level=WARN'); + expect(ws.url).toContain('source=sandbox'); + + act(() => ws.onopen?.()); + expect(result.current.isStreaming).toBe(true); + + act(() => { + ws.onmessage?.({ data: logFrame('first') }); + ws.onmessage?.({ data: logFrame('second') }); + // Non-log frames are ignored by this hook. + ws.onmessage?.({ + data: JSON.stringify({ type: 'warning', warning: 'x' }), + }); + }); + expect(result.current.lines.map((l) => l.message)).toEqual([ + 'first', + 'second', + ]); + }); + + it('caps the buffer at the requested line count', () => { + const { result } = renderHook(() => + useSandboxLogStream('default', 'sb1', { lines: 3 }, true), + ); + const ws = FakeWebSocket.instances[0]; + act(() => ws.onopen?.()); + act(() => { + for (let i = 1; i <= 5; i += 1) { + ws.onmessage?.({ data: logFrame(`line-${i}`) }); + } + }); + expect(result.current.lines.map((l) => l.message)).toEqual([ + 'line-3', + 'line-4', + 'line-5', + ]); + }); + + it('clears the buffer on reconnect so the tail replay is not duplicated', () => { + const { result } = renderHook(() => + useSandboxLogStream('default', 'sb1', { lines: 100 }, true), + ); + const ws = FakeWebSocket.instances[0]; + act(() => ws.onopen?.()); + act(() => ws.onmessage?.({ data: logFrame('old') })); + expect(result.current.lines).toHaveLength(1); + + act(() => ws.onclose?.()); + expect(result.current.isStreaming).toBe(false); + + act(() => jest.advanceTimersByTime(2_000)); + const reconnected = FakeWebSocket.instances[1]; + act(() => reconnected.onopen?.()); + expect(result.current.lines).toHaveLength(0); + + act(() => reconnected.onmessage?.({ data: logFrame('replayed') })); + expect(result.current.lines.map((l) => l.message)).toEqual(['replayed']); + }); + + it('closes the socket on unmount', () => { + const { unmount } = renderHook(() => + useSandboxLogStream('default', 'sb1', { lines: 100 }, true), + ); + const ws = FakeWebSocket.instances[0]; + unmount(); + expect(ws.closed).toBe(true); + }); +}); diff --git a/frontend/src/hooks/__tests__/useSandboxWatch.spec.tsx b/frontend/src/hooks/__tests__/useSandboxWatch.spec.tsx new file mode 100644 index 00000000..e60c5717 --- /dev/null +++ b/frontend/src/hooks/__tests__/useSandboxWatch.spec.tsx @@ -0,0 +1,158 @@ +import React from 'react'; +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import { act, renderHook } from '@testing-library/react'; + +import { useSandboxWatch } from '../useSandboxWatch'; +import { policyKeys, sandboxKeys } from '../../api/queryKeys'; + +const mockUseFeatureFlags = jest.fn(); +jest.mock('../../api/auth', () => ({ + useFeatureFlags: () => mockUseFeatureFlags(), +})); + +class FakeWebSocket { + static instances: FakeWebSocket[] = []; + url: string; + onopen: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + onclose: (() => void) | null = null; + onerror: (() => void) | null = null; + closed = false; + + constructor(url: string) { + this.url = url; + FakeWebSocket.instances.push(this); + } + + close() { + this.closed = true; + this.onclose?.(); + } +} + +describe('useSandboxWatch', () => { + let queryClient: QueryClient; + const realWebSocket = window.WebSocket; + + const wrapper = ({ children }: { children: React.ReactNode }) => ( + {children} + ); + + beforeEach(() => { + jest.useFakeTimers(); + queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + FakeWebSocket.instances = []; + (window as unknown as { WebSocket: unknown }).WebSocket = FakeWebSocket; + mockUseFeatureFlags.mockReturnValue({ liveUpdates: true }); + }); + + afterEach(() => { + jest.useRealTimers(); + (window as unknown as { WebSocket: unknown }).WebSocket = realWebSocket; + }); + + it('does not open a socket when the flag is off', () => { + mockUseFeatureFlags.mockReturnValue({ liveUpdates: false }); + const { result } = renderHook(() => useSandboxWatch('default', 'sb1'), { + wrapper, + }); + expect(FakeWebSocket.instances).toHaveLength(0); + expect(result.current.isLive).toBe(false); + }); + + it('opens a socket to the watch endpoint and reports live on open', () => { + const { result } = renderHook(() => useSandboxWatch('default', 'sb1'), { + wrapper, + }); + expect(FakeWebSocket.instances).toHaveLength(1); + expect(FakeWebSocket.instances[0].url).toContain( + '/api/v1/workspaces/default/sandboxes/sb1/watch', + ); + expect(result.current.isLive).toBe(false); + + act(() => FakeWebSocket.instances[0].onopen?.()); + expect(result.current.isLive).toBe(true); + }); + + it('pushes sandbox snapshots into the cache and invalidates draft queries', () => { + const invalidateSpy = jest.spyOn(queryClient, 'invalidateQueries'); + renderHook(() => useSandboxWatch('default', 'sb1'), { wrapper }); + const ws = FakeWebSocket.instances[0]; + act(() => ws.onopen?.()); + + const sandbox = { + metadata: { id: 'id-1', name: 'sb1', workspace: 'default' }, + spec: {}, + status: { phase: 'READY' }, + }; + act(() => + ws.onmessage?.({ data: JSON.stringify({ type: 'sandbox', sandbox }) }), + ); + + expect( + queryClient.getQueryData(sandboxKeys.detail('default', 'sb1')), + ).toEqual(sandbox); + + act(() => jest.advanceTimersByTime(300)); + const invalidatedKeys = invalidateSpy.mock.calls.map( + ([filters]) => filters?.queryKey, + ); + expect(invalidatedKeys).toContainEqual(policyKeys.drafts('default', 'sb1')); + expect(invalidatedKeys).toContainEqual( + policyKeys.sandbox('default', 'sb1'), + ); + expect(invalidatedKeys).toContainEqual( + policyKeys.draftHistory('default', 'sb1'), + ); + }); + + it('coalesces bursts of events into one invalidation round', () => { + const invalidateSpy = jest.spyOn(queryClient, 'invalidateQueries'); + renderHook(() => useSandboxWatch('default', 'sb1'), { wrapper }); + const ws = FakeWebSocket.instances[0]; + act(() => ws.onopen?.()); + + act(() => { + for (let i = 0; i < 5; i += 1) { + ws.onmessage?.({ + data: JSON.stringify({ type: 'warning', warning: 'lag' }), + }); + } + jest.advanceTimersByTime(300); + }); + + const draftInvalidations = invalidateSpy.mock.calls.filter( + ([filters]) => + JSON.stringify(filters?.queryKey) === + JSON.stringify(policyKeys.drafts('default', 'sb1')), + ); + expect(draftInvalidations).toHaveLength(1); + }); + + it('drops back to polling after the socket closes', () => { + const { result } = renderHook(() => useSandboxWatch('default', 'sb1'), { + wrapper, + }); + const ws = FakeWebSocket.instances[0]; + act(() => ws.onopen?.()); + expect(result.current.isLive).toBe(true); + + act(() => ws.onclose?.()); + expect(result.current.isLive).toBe(false); + + // A reconnect attempt is scheduled with backoff. + act(() => jest.advanceTimersByTime(2_000)); + expect(FakeWebSocket.instances.length).toBeGreaterThan(1); + }); + + it('closes the socket on unmount', () => { + const { unmount } = renderHook(() => useSandboxWatch('default', 'sb1'), { + wrapper, + }); + const ws = FakeWebSocket.instances[0]; + unmount(); + expect(ws.closed).toBe(true); + }); +}); diff --git a/frontend/src/hooks/useSandboxLogStream.ts b/frontend/src/hooks/useSandboxLogStream.ts new file mode 100644 index 00000000..c92d14c5 --- /dev/null +++ b/frontend/src/hooks/useSandboxLogStream.ts @@ -0,0 +1,78 @@ +import { useEffect, useMemo, useState } from 'react'; + +import { useFeatureFlags } from '../api/auth'; +import { openWatchSocket, watchSocketUrl } from './watchSocket'; +import type { LogLine } from '../types'; + +export type LogStreamFilters = { + lines: number; + level?: string; + sources?: string[]; +}; + +/** + * Streams sandbox log lines over the watch WebSocket (follow_logs): the BFF + * replays the last `lines` as a tail, then pushes live lines. The buffer is + * capped at `lines` so it mirrors the polled view's line-count selector. + * + * Returns isStreaming so the caller can suspend its polled logs query while + * the stream is healthy — like useSandboxWatch, polling is the fallback. + * Disabled entirely (no socket) when `enabled` is false or the liveUpdates + * feature flag is off. + */ +export const useSandboxLogStream = ( + workspace: string, + name: string, + filters: LogStreamFilters, + enabled: boolean, +): { lines: LogLine[]; isStreaming: boolean } => { + const flags = useFeatureFlags(); + const [lines, setLines] = useState([]); + const [isStreaming, setIsStreaming] = useState(false); + const active = enabled && !!flags.liveUpdates; + + const { lines: maxLines, level } = filters; + const sourcesKey = (filters.sources ?? []).join(','); + + const params = useMemo(() => { + const search = new URLSearchParams({ logs: 'true' }); + search.set('lines', String(maxLines)); + if (level) search.set('level', level); + for (const source of sourcesKey.split(',')) { + if (source) search.append('source', source); + } + return search.toString(); + }, [maxLines, level, sourcesKey]); + + useEffect(() => { + if (!active) return undefined; + setLines([]); + + const handle = openWatchSocket({ + url: watchSocketUrl(workspace, name, new URLSearchParams(params)), + onEvent: (event) => { + if (event.type !== 'log' || !event.log) return; + const line = event.log; + setLines((prev) => { + const next = [...prev, line]; + return next.length > maxLines + ? next.slice(next.length - maxLines) + : next; + }); + }, + onLiveChange: (live) => { + // Each (re)connect replays the tail, so reset the buffer to avoid + // duplicating replayed lines. + if (live) setLines([]); + setIsStreaming(live); + }, + }); + + return () => { + handle.dispose(); + setIsStreaming(false); + }; + }, [active, workspace, name, params, maxLines]); + + return { lines, isStreaming }; +}; diff --git a/frontend/src/hooks/useSandboxWatch.ts b/frontend/src/hooks/useSandboxWatch.ts new file mode 100644 index 00000000..6969da7f --- /dev/null +++ b/frontend/src/hooks/useSandboxWatch.ts @@ -0,0 +1,82 @@ +import { useEffect, useState } from 'react'; +import { useQueryClient } from '@tanstack/react-query'; + +import { useFeatureFlags } from '../api/auth'; +import { policyKeys, sandboxKeys } from '../api/queryKeys'; +import { openWatchSocket, watchSocketUrl } from './watchSocket'; + +export type { SandboxWatchEvent } from './watchSocket'; + +// Gateway watch-bus events can burst (one per draft mutation); coalesce +// refetches into one. +const INVALIDATE_DEBOUNCE_MS = 250; + +/** + * Subscribes to the BFF's sandbox watch WebSocket (a relay of the gateway's + * WatchSandbox stream) and translates events into React Query cache updates: + * sandbox snapshots are pushed directly into the detail cache, and draft / + * policy queries are invalidated so they refetch. + * + * Returns isLive so callers can switch their queries' refetchInterval off + * while the socket is open — polling is the fallback, not a parallel system. + * When the liveUpdates feature flag is off, no socket is opened and isLive + * stays false. + */ +export const useSandboxWatch = ( + workspace: string, + name: string, +): { isLive: boolean } => { + const flags = useFeatureFlags(); + const queryClient = useQueryClient(); + const [isLive, setIsLive] = useState(false); + const enabled = !!flags.liveUpdates; + + useEffect(() => { + if (!enabled) return undefined; + + let invalidateTimer: number | undefined; + + const invalidateSoon = () => { + if (invalidateTimer !== undefined) return; + invalidateTimer = window.setTimeout(() => { + invalidateTimer = undefined; + queryClient.invalidateQueries({ + queryKey: policyKeys.drafts(workspace, name), + }); + queryClient.invalidateQueries({ + queryKey: policyKeys.sandbox(workspace, name), + }); + queryClient.invalidateQueries({ + queryKey: policyKeys.draftHistory(workspace, name), + }); + }, INVALIDATE_DEBOUNCE_MS); + }; + + const handle = openWatchSocket({ + url: watchSocketUrl(workspace, name), + onEvent: (event) => { + if (event.type === 'sandbox' && event.sandbox) { + queryClient.setQueryData( + sandboxKeys.detail(workspace, name), + event.sandbox, + ); + invalidateSoon(); + } else if ( + event.type === 'draftPolicyUpdate' || + event.type === 'warning' + ) { + invalidateSoon(); + } + }, + onLiveChange: setIsLive, + }); + + return () => { + if (invalidateTimer !== undefined) window.clearTimeout(invalidateTimer); + handle.dispose(); + setIsLive(false); + }; + }, [enabled, workspace, name, queryClient]); + + return { isLive }; +}; diff --git a/frontend/src/hooks/watchSocket.ts b/frontend/src/hooks/watchSocket.ts new file mode 100644 index 00000000..ff77e84b --- /dev/null +++ b/frontend/src/hooks/watchSocket.ts @@ -0,0 +1,105 @@ +import type { LogLine, Sandbox } from '../types'; + +// One JSON frame from the BFF's sandbox watch WebSocket (a relay of the +// gateway's WatchSandbox stream — see backend models.WatchEvent). +export type SandboxWatchEvent = { + type: 'sandbox' | 'log' | 'warning' | 'draftPolicyUpdate'; + sandbox?: Sandbox; + log?: LogLine; + warning?: string; + draftPolicyUpdate?: { + draftVersion: number; + newChunks: number; + totalPending: number; + }; +}; + +const RECONNECT_BASE_MS = 1_000; +const RECONNECT_MAX_MS = 30_000; +// After this many failed connects in a row, stop retrying — the proxy likely +// cannot upgrade WebSockets and polling has already taken over. +const MAX_CONSECUTIVE_FAILURES = 5; + +export const watchSocketUrl = ( + workspace: string, + name: string, + params?: URLSearchParams, +): string => { + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const search = params?.toString() ?? ''; + const query = search ? `?${search}` : ''; + return `${protocol}//${window.location.host}/api/v1/workspaces/${encodeURIComponent(workspace)}/sandboxes/${encodeURIComponent(name)}/watch${query}`; +}; + +export type WatchSocketHandle = { + dispose: () => void; +}; + +type WatchSocketOptions = { + url: string; + onEvent: (event: SandboxWatchEvent) => void; + onLiveChange: (live: boolean) => void; +}; + +/** + * Opens a WebSocket to `url` and keeps it alive: reconnects with capped, + * jittered exponential backoff and gives up after repeated consecutive + * failures. onLiveChange(true/false) tracks connection state so callers can + * toggle their polling fallback. dispose() tears everything down. + */ +export const openWatchSocket = ({ + url, + onEvent, + onLiveChange, +}: WatchSocketOptions): WatchSocketHandle => { + let ws: WebSocket | null = null; + let disposed = false; + let failures = 0; + let reconnectTimer: number | undefined; + + const connect = () => { + ws = new WebSocket(url); + + ws.onopen = () => { + failures = 0; + onLiveChange(true); + }; + + ws.onmessage = (event) => { + let parsed: SandboxWatchEvent; + try { + parsed = JSON.parse(event.data as string) as SandboxWatchEvent; + } catch { + return; + } + onEvent(parsed); + }; + + ws.onclose = () => { + onLiveChange(false); + if (disposed) return; + failures += 1; + if (failures > MAX_CONSECUTIVE_FAILURES) return; + const backoff = Math.min( + RECONNECT_BASE_MS * 2 ** (failures - 1), + RECONNECT_MAX_MS, + ); + const delay = backoff * (0.5 + Math.random() * 0.5); + reconnectTimer = window.setTimeout(connect, delay); + }; + + ws.onerror = () => { + ws?.close(); + }; + }; + + connect(); + + return { + dispose: () => { + disposed = true; + if (reconnectTimer !== undefined) window.clearTimeout(reconnectTimer); + ws?.close(); + }, + }; +}; diff --git a/frontend/src/pages/SandboxDetailPage.tsx b/frontend/src/pages/SandboxDetailPage.tsx index 500a753a..4746d798 100644 --- a/frontend/src/pages/SandboxDetailPage.tsx +++ b/frontend/src/pages/SandboxDetailPage.tsx @@ -40,6 +40,7 @@ import SandboxTerminalTab from '../components/sandbox/SandboxTerminalTab'; import PolicyRuleEditor from '../components/PolicyRuleEditor'; import SandboxProvidersTab from '../components/sandbox/SandboxProvidersTab'; import SandboxServicesTab from '../components/sandbox/SandboxServicesTab'; +import { useSandboxWatch } from '../hooks/useSandboxWatch'; import { formatTimestamp } from '../utils/formatters'; type SandboxDetailPageProps = { @@ -51,10 +52,13 @@ type SandboxDetailPageProps = { const SandboxDetailPage: React.FC = (props) => { const { workspace, sandboxName } = props; - const sandbox = useSandbox(workspace, sandboxName); + const { isLive } = useSandboxWatch(workspace, sandboxName); + const sandbox = useSandbox(workspace, sandboxName, { live: isLive }); const features = useFeatureFlags(); const policyQuery = useSandboxPolicy(workspace, sandboxName); - const draftsQuery = useDraftPolicy(workspace, sandboxName); + const draftsQuery = useDraftPolicy(workspace, sandboxName, { + live: isLive, + }); const providerExpiry = useProviderExpiry(workspace); const [searchParams, setSearchParams] = useSearchParams(); const resolvedTab = props.activeTab ?? (searchParams.get('tab') || 'details'); @@ -126,6 +130,17 @@ const SandboxDetailPage: React.FC = (props) => { + {features.liveUpdates && ( + + + + )} = (props) => { diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 5c26415c..e8f0706d 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -316,6 +316,7 @@ export type FeatureFlags = { credentialRefresh: boolean; services: boolean; draftPolicy: boolean; + liveUpdates: boolean; }; export type AuthConfig = {