From 50e0390a8157250ee9613233cc101b1d1940ea93 Mon Sep 17 00:00:00 2001 From: tmphost-dev Date: Fri, 23 Oct 2026 04:23:31 +0000 Subject: [PATCH] Stream query activity live over SSE, with a query detail sidebar The Activity tab now subscribes to GET /api/v1/databases/{id}/queries/stream and adds queries as they are recorded. Clicking a row opens a right-hand sidebar with the full SQL, its status, reason, rows, duration and result. The stream is served by the stdlib mux, since the Fiber adaptor buffers whole responses. It ends with the session or the database; the browser renews the session and reconnects, catching up from /queries. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 3 + backend/internal/app/app.go | 2 + backend/internal/app/app_test.go | 33 +++ backend/internal/controller/crud/crud.go | 10 +- backend/internal/controller/crud/crud_test.go | 46 ++++ backend/internal/controller/crud/query.go | 32 ++- backend/internal/data/view/view.go | 12 +- backend/internal/handler/live/live.go | 137 +++++++++++ backend/internal/handler/live/live_test.go | 218 ++++++++++++++++++ backend/internal/service/feed/feed.go | 66 ++++++ backend/internal/service/feed/feed_test.go | 53 +++++ backend/internal/testutil/testutil.go | 6 +- docs/API.md | 9 + frontend/src/api.js | 10 + frontend/src/components/ActivityPanel.vue | 155 ++++++++----- frontend/src/components/QueryDetail.vue | 75 ++++++ frontend/src/components/queryStatus.js | 6 + frontend/src/composables/useFormat.js | 8 + frontend/src/composables/useQueryStream.js | 105 +++++++++ frontend/test/ActivityPanel.test.js | 138 +++++++++++ frontend/test/api.test.js | 9 + frontend/test/composables.test.js | 5 +- frontend/test/useQueryStream.test.js | 159 +++++++++++++ 23 files changed, 1230 insertions(+), 67 deletions(-) create mode 100644 backend/internal/handler/live/live.go create mode 100644 backend/internal/handler/live/live_test.go create mode 100644 backend/internal/service/feed/feed.go create mode 100644 backend/internal/service/feed/feed_test.go create mode 100644 frontend/src/components/QueryDetail.vue create mode 100644 frontend/src/components/queryStatus.js create mode 100644 frontend/src/composables/useQueryStream.js create mode 100644 frontend/test/ActivityPanel.test.js create mode 100644 frontend/test/useQueryStream.test.js diff --git a/AGENTS.md b/AGENTS.md index f56cd4e..d74dde2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -45,6 +45,9 @@ app state). A handler never calls the repository directly. - **OAuth registrations, consents and codes are signed tokens, not rows.** Deleting a database revokes its grants because every token is checked against the database's current owner. +- **Streams go on the stdlib mux, not Fiber** (`handler/live`). The Fiber + adaptor buffers the whole response, so an SSE route on Fiber sends nothing + until it ends. The live query feed (`service/feed`) is in-process only. - **Vite builds with `base: ''`.** The server rewrites `./` to `/` in `index.html` (`app.spa`). Without that rewrite, a hard reload on a nested route such as `/databases/x` serves a blank page. diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 53a475c..20f3e58 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -23,6 +23,7 @@ import ( "github.com/agentrq/sql-mcp/backend/internal/controller/mcpserver" "github.com/agentrq/sql-mcp/backend/internal/handler/admin" "github.com/agentrq/sql-mcp/backend/internal/handler/api" + "github.com/agentrq/sql-mcp/backend/internal/handler/live" mcphandler "github.com/agentrq/sql-mcp/backend/internal/handler/mcp" "github.com/agentrq/sql-mcp/backend/internal/repository" "github.com/agentrq/sql-mcp/backend/internal/service/auth" @@ -134,6 +135,7 @@ func New(cfg config.Config, deps Deps) (http.Handler, error) { MCP: mcpserver.NewHandler(ctrl, Version), BaseURL: cfg.BaseURL, }) + live.Register(live.Params{Mux: mux, Watcher: ctrl, Tokens: tokens}) mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"status":"ok"}`)) diff --git a/backend/internal/app/app_test.go b/backend/internal/app/app_test.go index 2409c45..b44e280 100644 --- a/backend/internal/app/app_test.go +++ b/backend/internal/app/app_test.go @@ -1,6 +1,7 @@ package app_test import ( + "bufio" "context" "crypto/sha256" "encoding/base64" @@ -623,3 +624,35 @@ func TestNewRefusesAnUnreadableSavedConfiguration(t *testing.T) { _, err = app.New(cfg, app.Deps{}) assert.ErrorContains(t, err, "ENCRYPTION_KEY") } + +func TestQueryStream(t *testing.T) { + s := newStack(t) + c := s.rootSession() + db := "/api/v1/databases/" + s.createDB(c) + assert.Equal(t, http.StatusNotFound, s.do(s.googleSession("bob@example.com"), "GET", db+"/queries/stream", nil).status) + assert.Equal(t, http.StatusUnauthorized, s.do(s.browser(), "GET", db+"/queries/stream", nil).status) + + res, err := c.Get(s.srv.URL + db + "/queries/stream") + require.NoError(t, err) + defer res.Body.Close() + require.Equal(t, http.StatusOK, res.StatusCode) + assert.Equal(t, "text/event-stream", res.Header.Get("Content-Type")) + r := bufio.NewReader(res.Body) + line, err := r.ReadString('\n') + require.NoError(t, err) + assert.Equal(t, ": connected\n", line, "the stream is flushed, not buffered by Fiber") + + s.do(c, "POST", db+"/query", map[string]string{"sql": "select id from users"}) + var event strings.Builder + for !strings.Contains(event.String(), "data: ") { + line, err = r.ReadString('\n') + require.NoError(t, err) + event.WriteString(line) + } + assert.Contains(t, event.String(), "event: query\n") + assert.Contains(t, event.String(), `"sql":"select id from users"`) + + assert.Equal(t, http.StatusNoContent, s.do(c, "DELETE", db, nil).status) + _, err = io.ReadAll(r) + assert.NoError(t, err, "deleting the database ends the stream") +} diff --git a/backend/internal/controller/crud/crud.go b/backend/internal/controller/crud/crud.go index bb382cf..a41a572 100644 --- a/backend/internal/controller/crud/crud.go +++ b/backend/internal/controller/crud/crud.go @@ -18,6 +18,7 @@ import ( "github.com/agentrq/sql-mcp/backend/internal/repository" "github.com/agentrq/sql-mcp/backend/internal/service/auth" "github.com/agentrq/sql-mcp/backend/internal/service/dbdriver" + "github.com/agentrq/sql-mcp/backend/internal/service/feed" "github.com/agentrq/sql-mcp/backend/internal/service/security" "github.com/agentrq/sql-mcp/backend/internal/service/tmphost" ) @@ -50,6 +51,7 @@ type Controller struct { tmphost *tmphost.Client cfg Config current atomic.Pointer[live] + queries *feed.Hub[entity.QueryLog] // recorded queries, keyed by database ID now func() time.Time } @@ -63,6 +65,7 @@ func New(p Params) *Controller { envAllowlist: p.Allowlist, tmphost: p.Tmphost, cfg: p.Config, + queries: feed.New[entity.QueryLog](), now: time.Now, } c.current.Store(c.envSetup()) @@ -271,7 +274,12 @@ func (c *Controller) UpdateDatabase(ctx context.Context, userID, id int64, name, // DeleteDatabase removes a database; its MCP tokens stop working with it. func (c *Controller) DeleteDatabase(ctx context.Context, userID, id int64) error { - return mapErr(c.repo.DeleteDatabase(ctx, userID, id)) + if err := c.repo.DeleteDatabase(ctx, userID, id); err != nil { + return mapErr(err) + } + // Whoever is watching its activity loses access with the database. + c.queries.Close(id) + return nil } // open returns a person's database with its engine and opened DSN. diff --git a/backend/internal/controller/crud/crud_test.go b/backend/internal/controller/crud/crud_test.go index f163da1..e38546c 100644 --- a/backend/internal/controller/crud/crud_test.go +++ b/backend/internal/controller/crud/crud_test.go @@ -9,8 +9,10 @@ import ( "testing" "time" + "github.com/glebarez/sqlite" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "gorm.io/gorm" "github.com/agentrq/sql-mcp/backend/internal/controller/crud" "github.com/agentrq/sql-mcp/backend/internal/data/entity" @@ -389,3 +391,47 @@ func TestToCSV(t *testing.T) { require.NoError(t, err) assert.Contains(t, string(out), "0x") } + +func TestWatchQueries(t *testing.T) { + env := testutil.New(t) + alice := signIn(t, env, "alice@example.com") + bob := signIn(t, env, "bob@example.com") + id := newDB(t, env, alice) + + _, _, err := env.Ctrl.WatchQueries(ctx, bob, id) + assert.ErrorIs(t, err, entity.ErrNotFound) + + queries, stop, err := env.Ctrl.WatchQueries(ctx, alice, id) + require.NoError(t, err) + _, err = env.Ctrl.RunQuery(ctx, crud.QueryRequest{UserID: alice, DatabaseID: id, SQL: "select id from users", Source: model.SourceMCP}) + require.NoError(t, err) + q := <-queries + logs, err := env.Ctrl.ListQueries(ctx, alice, id, 1) + require.NoError(t, err) + assert.Equal(t, logs[0].ID, q.ID, "a watcher sees what the log lists") + assert.True(t, logs[0].CreatedAt.Equal(q.CreatedAt)) + assert.Equal(t, "select id from users", q.SQL) + assert.Equal(t, model.QueryOK, q.Status) + + require.NoError(t, env.Ctrl.DeleteDatabase(ctx, alice, id)) + _, open := <-queries + assert.False(t, open, "deleting the database ends the watch") + stop() +} + +func TestAQueryThatCannotBeLoggedIsNotWatched(t *testing.T) { + env := testutil.New(t) + alice := signIn(t, env, "alice@example.com") + id := newDB(t, env, alice) + queries, stop, err := env.Ctrl.WatchQueries(ctx, alice, id) + require.NoError(t, err) + defer stop() + + db, err := gorm.Open(sqlite.Open(env.DBPath), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.Exec("DROP TABLE query_logs").Error) + + _, err = env.Ctrl.RunQuery(ctx, crud.QueryRequest{UserID: alice, DatabaseID: id, SQL: "select id from users", Source: model.SourceConsole}) + require.NoError(t, err, "a failure to log does not fail the query") + assert.Empty(t, queries) +} diff --git a/backend/internal/controller/crud/query.go b/backend/internal/controller/crud/query.go index 9707392..0a44439 100644 --- a/backend/internal/controller/crud/query.go +++ b/backend/internal/controller/crud/query.go @@ -88,12 +88,15 @@ func (c *Controller) RunQuery(ctx context.Context, rq QueryRequest) (*entity.Que return out, nil } -// record writes the query log. A failure to log does not fail the query the -// person already has, so it is only reported. +// record writes the query log and tells whoever watches the database. A +// failure to log does not fail the query the person already has, so it is +// only reported. func (c *Controller) record(ctx context.Context, q *model.QueryLog) { if err := c.repo.AddQueryLog(context.WithoutCancel(ctx), q); err != nil { log.Error().Err(err).Msg("could not record a query") + return } + c.queries.Publish(q.DatabaseID, queryLogEntity(q)) } func truncate(s string, n int) string { @@ -116,15 +119,30 @@ func (c *Controller) ListQueries(ctx context.Context, userID, id int64, limit in return nil, err } out := make([]entity.QueryLog, 0, len(rows)) - for _, q := range rows { - out = append(out, entity.QueryLog{ - ID: FormatID(q.ID), Source: q.Source, SQL: q.SQL, Status: q.Status, Reason: q.Reason, - RowCount: q.RowCount, ResultURL: q.ResultURL, DurationMs: q.DurationMs, CreatedAt: q.CreatedAt, - }) + for i := range rows { + out = append(out, queryLogEntity(&rows[i])) } return out, nil } +// WatchQueries streams a database's queries as they are recorded. The +// channel is closed when stop is called or the database is deleted; a +// watcher that falls behind misses queries rather than slowing them down. +func (c *Controller) WatchQueries(ctx context.Context, userID, id int64) (queries <-chan entity.QueryLog, stop func(), err error) { + if _, err := c.repo.GetDatabase(ctx, userID, id); err != nil { + return nil, nil, mapErr(err) + } + queries, stop = c.queries.Subscribe(id, 64) + return queries, stop, nil +} + +func queryLogEntity(q *model.QueryLog) entity.QueryLog { + return entity.QueryLog{ + ID: FormatID(q.ID), Source: q.Source, SQL: q.SQL, Status: q.Status, Reason: q.Reason, + RowCount: q.RowCount, ResultURL: q.ResultURL, DurationMs: q.DurationMs, CreatedAt: q.CreatedAt, + } +} + // SystemDatabaseOwner returns who owns a database and its name, for the MCP // server to check a token against before anything else. func (c *Controller) SystemDatabaseOwner(ctx context.Context, id int64) (ownerID int64, name string, err error) { diff --git a/backend/internal/data/view/view.go b/backend/internal/data/view/view.go index ae17ffe..11ea9d8 100644 --- a/backend/internal/data/view/view.go +++ b/backend/internal/data/view/view.go @@ -221,14 +221,18 @@ func FromQueryResult(r *entity.QueryResult) QueryResponse { func FromQueryLogs(in []entity.QueryLog) QueryLogsResponse { out := QueryLogsResponse{Queries: []QueryLog{}} for _, q := range in { - out.Queries = append(out.Queries, QueryLog{ - ID: q.ID, Source: q.Source, SQL: q.SQL, Status: q.Status, Reason: q.Reason, - RowCount: q.RowCount, ResultURL: q.ResultURL, DurationMs: q.DurationMs, CreatedAt: q.CreatedAt, - }) + out.Queries = append(out.Queries, FromQueryLog(q)) } return out } +func FromQueryLog(q entity.QueryLog) QueryLog { + return QueryLog{ + ID: q.ID, Source: q.Source, SQL: q.SQL, Status: q.Status, Reason: q.Reason, + RowCount: q.RowCount, ResultURL: q.ResultURL, DurationMs: q.DurationMs, CreatedAt: q.CreatedAt, + } +} + func FromTmphostStatus(s *entity.TmphostStatus) TmphostStatus { out := TmphostStatus{Connected: s.Connected, Account: s.Account, ServerURL: s.ServerURL} if s.Connected { diff --git a/backend/internal/handler/live/live.go b/backend/internal/handler/live/live.go new file mode 100644 index 0000000..e4be954 --- /dev/null +++ b/backend/internal/handler/live/live.go @@ -0,0 +1,137 @@ +// Package live streams a database's queries to the browser as server-sent +// events. It sits on the standard mux, not Fiber: Fiber reaches net/http +// through an adaptor that buffers the whole response, so nothing would be +// sent until the stream ended. +package live + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "time" + + "github.com/rs/zerolog/log" + + "github.com/agentrq/sql-mcp/backend/internal/controller/crud" + "github.com/agentrq/sql-mcp/backend/internal/data/entity" + "github.com/agentrq/sql-mcp/backend/internal/data/view" + "github.com/agentrq/sql-mcp/backend/internal/service/auth" +) + +// accessCookie is the session cookie the REST API sets. +const accessCookie = "at" + +// Watcher streams a database's queries. +type Watcher interface { + WatchQueries(ctx context.Context, userID, id int64) (<-chan entity.QueryLog, func(), error) +} + +// Params is what the handler needs. +type Params struct { + Mux *http.ServeMux + Watcher Watcher + Tokens *auth.TokenService + // Heartbeat is how often an idle stream sends a comment, so proxies do + // not close it. Zero is 25 seconds. + Heartbeat time.Duration +} + +type handler struct { + watcher Watcher + tokens *auth.TokenService + heartbeat time.Duration +} + +// Register mounts the stream. The pattern is more specific than the API's +// `/api/v1/`, so the mux sends it here rather than to Fiber. +func Register(p Params) { + h := &handler{watcher: p.Watcher, tokens: p.Tokens, heartbeat: p.Heartbeat} + if h.heartbeat == 0 { + h.heartbeat = 25 * time.Second + } + p.Mux.HandleFunc("GET /api/v1/databases/{id}/queries/stream", h.queries) +} + +func fail(w http.ResponseWriter, status int, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(view.Error{Error: view.ErrorBody{Message: message, Code: status}}) +} + +// queries sends `event: query` with a query log entry as each one is +// recorded. It ends when the session expires, so a stream never outlives the +// cookie that opened it, and when the database is deleted; the browser +// reconnects, and is refused, either way. +func (h *handler) queries(w http.ResponseWriter, r *http.Request) { + var userID int64 + var expires time.Time + cookie, err := r.Cookie(accessCookie) + if err == nil { + if claims, err := h.tokens.ValidateAccessToken(cookie.Value); err == nil { + userID, expires = crud.ParseID(claims.Subject), claims.ExpiresAt.Time + } + } + if userID == 0 { + fail(w, http.StatusUnauthorized, "unauthorized") + return + } + + queries, stop, err := h.watcher.WatchQueries(r.Context(), userID, crud.ParseID(r.PathValue("id"))) + if errors.Is(err, entity.ErrNotFound) { + fail(w, http.StatusNotFound, "not found") + return + } + if err != nil { + log.Error().Err(err).Str("path", r.URL.Path).Msg("request failed") + fail(w, http.StatusInternalServerError, "internal server error") + return + } + defer stop() + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("X-Accel-Buffering", "no") // nginx would hold events back + w.WriteHeader(http.StatusOK) + rc := http.NewResponseController(w) + // The comment opens the stream at once, so the browser reports it open + // before the first query arrives. + if !send(w, rc, ": connected\n\n") { + return + } + + session := time.NewTimer(time.Until(expires)) + defer session.Stop() + beat := time.NewTicker(h.heartbeat) + defer beat.Stop() + for { + var msg string + select { + case <-r.Context().Done(): + return + case <-session.C: + return + case q, open := <-queries: + if !open { + return + } + data, _ := json.Marshal(view.FromQueryLog(q)) + msg = fmt.Sprintf("id: %s\nevent: query\ndata: %s\n\n", q.ID, data) + case <-beat.C: + msg = ": ping\n\n" + } + if !send(w, rc, msg) { + return + } + } +} + +// send writes msg and flushes it, reporting whether the client is still +// there. +func send(w http.ResponseWriter, rc *http.ResponseController, msg string) bool { + if _, err := w.Write([]byte(msg)); err != nil { + return false + } + return rc.Flush() == nil +} diff --git a/backend/internal/handler/live/live_test.go b/backend/internal/handler/live/live_test.go new file mode 100644 index 0000000..5b3c7e0 --- /dev/null +++ b/backend/internal/handler/live/live_test.go @@ -0,0 +1,218 @@ +package live_test + +import ( + "bufio" + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/agentrq/sql-mcp/backend/internal/controller/crud" + "github.com/agentrq/sql-mcp/backend/internal/data/entity" + "github.com/agentrq/sql-mcp/backend/internal/handler/live" + "github.com/agentrq/sql-mcp/backend/internal/service/auth" + "github.com/agentrq/sql-mcp/backend/internal/testutil" +) + +const ( + userID = int64(1_000_001) + dbID = int64(1_000_002) +) + +// watcher hands out ch to userID for dbID, and fails otherwise. +type watcher struct { + ch chan entity.QueryLog + err error + stopped chan struct{} +} + +func (w *watcher) WatchQueries(_ context.Context, uid, id int64) (<-chan entity.QueryLog, func(), error) { + if w.err != nil { + return nil, nil, w.err + } + if uid != userID || id != dbID { + return nil, nil, entity.ErrNotFound + } + return w.ch, func() { close(w.stopped) }, nil +} + +func newWatcher() *watcher { + return &watcher{ch: make(chan entity.QueryLog, 4), stopped: make(chan struct{})} +} + +func tokens(t *testing.T) *auth.TokenService { + t.Helper() + s, err := auth.NewTokenService(testutil.Secret) + require.NoError(t, err) + return s +} + +func session(t *testing.T) string { + t.Helper() + tok, err := tokens(t).CreateAccessToken(crud.FormatID(userID), "a@example.com", "A", "") + require.NoError(t, err) + return tok +} + +func server(t *testing.T, w live.Watcher, heartbeat time.Duration) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + live.Register(live.Params{Mux: mux, Watcher: w, Tokens: tokens(t), Heartbeat: heartbeat}) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv +} + +func open(t *testing.T, srv *httptest.Server, id int64, token string) *http.Response { + t.Helper() + req, err := http.NewRequest("GET", srv.URL+"/api/v1/databases/"+crud.FormatID(id)+"/queries/stream", nil) + require.NoError(t, err) + if token != "" { + req.AddCookie(&http.Cookie{Name: "at", Value: token}) + } + res, err := srv.Client().Do(req) + require.NoError(t, err) + t.Cleanup(func() { _ = res.Body.Close() }) + return res +} + +// next reads one event (up to its blank line) from the stream. +func next(t *testing.T, r *bufio.Reader) string { + t.Helper() + var b strings.Builder + for { + line, err := r.ReadString('\n') + require.NoError(t, err) + if line == "\n" { + return b.String() + } + b.WriteString(line) + } +} + +func TestStreamRefusesWithoutAValidSessionOrDatabase(t *testing.T) { + srv := server(t, newWatcher(), 0) + assert.Equal(t, http.StatusUnauthorized, open(t, srv, dbID, "").StatusCode) + assert.Equal(t, http.StatusUnauthorized, open(t, srv, dbID, "not-a-jwt").StatusCode) + admin, _, err := tokens(t).CreateAdminToken() + require.NoError(t, err) + assert.Equal(t, http.StatusUnauthorized, open(t, srv, dbID, admin).StatusCode, "an admin token is not a session") + + res := open(t, srv, dbID+1, session(t)) + assert.Equal(t, http.StatusNotFound, res.StatusCode) + assert.Equal(t, "application/json", res.Header.Get("Content-Type")) + + broken := newWatcher() + broken.err = errors.New("db down") + res = open(t, server(t, broken, 0), dbID, session(t)) + assert.Equal(t, http.StatusInternalServerError, res.StatusCode) +} + +func TestStreamSendsQueriesAndHeartbeats(t *testing.T) { + w := newWatcher() + res := open(t, server(t, w, 20*time.Millisecond), dbID, session(t)) + require.Equal(t, http.StatusOK, res.StatusCode) + assert.Equal(t, "text/event-stream", res.Header.Get("Content-Type")) + r := bufio.NewReader(res.Body) + assert.Equal(t, ": connected\n", next(t, r)) + + w.ch <- entity.QueryLog{ID: "q1", SQL: "select 1", Status: "ok"} + ev := next(t, r) + for ev == ": ping\n" { + ev = next(t, r) + } + assert.True(t, strings.HasPrefix(ev, "id: q1\nevent: query\ndata: {"), ev) + assert.Contains(t, ev, `"sql":"select 1"`) + assert.Equal(t, ": ping\n", next(t, r)) + + close(w.ch) // the database was deleted + _, err := r.ReadString('\n') + assert.Error(t, err) + <-w.stopped +} + +func TestStreamEndsWithTheSession(t *testing.T) { + w := newWatcher() + claims := auth.Claims{RegisteredClaims: jwt.RegisteredClaims{ + Subject: crud.FormatID(userID), + Audience: jwt.ClaimStrings{auth.AccessAudience}, + ExpiresAt: jwt.NewNumericDate(time.Now().Add(2 * time.Second)), + }} + short, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(testutil.Secret)) + require.NoError(t, err) + + res := open(t, server(t, w, time.Hour), dbID, short) + require.Equal(t, http.StatusOK, res.StatusCode) + r := bufio.NewReader(res.Body) + next(t, r) + _, err = r.ReadString('\n') + assert.Error(t, err) + <-w.stopped +} + +func TestStreamStopsWhenTheClientLeaves(t *testing.T) { + w := newWatcher() + srv := server(t, w, time.Hour) + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, "GET", srv.URL+"/api/v1/databases/"+crud.FormatID(dbID)+"/queries/stream", nil) + req.AddCookie(&http.Cookie{Name: "at", Value: session(t)}) + res, err := srv.Client().Do(req) + require.NoError(t, err) + defer res.Body.Close() + next(t, bufio.NewReader(res.Body)) + cancel() + <-w.stopped +} + +// unflushable is a writer the stream cannot flush, like a proxy that buffers. +type unflushable struct{ header http.Header } + +func (u *unflushable) Header() http.Header { return u.header } +func (u *unflushable) Write(b []byte) (int, error) { return len(b), nil } +func (u *unflushable) WriteHeader(int) {} + +// failing accepts n writes, then fails, like a client that went away. +type failing struct { + *httptest.ResponseRecorder + n int +} + +func (f *failing) Write(b []byte) (int, error) { + if f.n == 0 { + return 0, errors.New("broken pipe") + } + f.n-- + return f.ResponseRecorder.Write(b) +} + +func TestStreamStopsWhenItCannotWrite(t *testing.T) { + w := newWatcher() + mux := http.NewServeMux() + live.Register(live.Params{Mux: mux, Watcher: w, Tokens: tokens(t), Heartbeat: time.Millisecond}) + req := httptest.NewRequest("GET", "/api/v1/databases/"+crud.FormatID(dbID)+"/queries/stream", nil) + req.AddCookie(&http.Cookie{Name: "at", Value: session(t)}) + + mux.ServeHTTP(&unflushable{header: http.Header{}}, req) + <-w.stopped + + for _, n := range []int{0, 1} { // the opening comment, then a heartbeat + w = newWatcher() + mux = http.NewServeMux() + live.Register(live.Params{Mux: mux, Watcher: w, Tokens: tokens(t), Heartbeat: time.Millisecond}) + mux.ServeHTTP(&failing{ResponseRecorder: httptest.NewRecorder(), n: n}, req) + <-w.stopped + } +} + +func TestStreamDefaultsItsHeartbeat(t *testing.T) { + w := newWatcher() + res := open(t, server(t, w, 0), dbID, session(t)) + require.Equal(t, http.StatusOK, res.StatusCode) +} diff --git a/backend/internal/service/feed/feed.go b/backend/internal/service/feed/feed.go new file mode 100644 index 0000000..1639c6b --- /dev/null +++ b/backend/internal/service/feed/feed.go @@ -0,0 +1,66 @@ +// Package feed fans events out to live subscribers, grouped by key. It is +// in-process: a subscriber sees what this server publishes, nothing else. +package feed + +import "sync" + +// Hub delivers each event published under a key to that key's subscribers. +// A subscriber that falls behind loses events rather than holding up the +// publisher, which is a query that has already run. +type Hub[T any] struct { + mu sync.Mutex + subs map[int64]map[chan T]struct{} +} + +// New is an empty hub. +func New[T any]() *Hub[T] { + return &Hub[T]{subs: map[int64]map[chan T]struct{}{}} +} + +// Subscribe returns key's events, buffering up to buffer of them, and the +// function that ends the subscription. The channel is closed when it ends, +// by that function or by Close. +func (h *Hub[T]) Subscribe(key int64, buffer int) (<-chan T, func()) { + ch := make(chan T, buffer) + h.mu.Lock() + if h.subs[key] == nil { + h.subs[key] = map[chan T]struct{}{} + } + h.subs[key][ch] = struct{}{} + h.mu.Unlock() + return ch, func() { + h.mu.Lock() + defer h.mu.Unlock() + // Close may have ended it already; a channel is closed only by + // whoever removes it, so never twice. + if _, ok := h.subs[key][ch]; ok { + delete(h.subs[key], ch) + if len(h.subs[key]) == 0 { + delete(h.subs, key) + } + close(ch) + } + } +} + +// Publish sends v to key's subscribers, skipping any whose buffer is full. +func (h *Hub[T]) Publish(key int64, v T) { + h.mu.Lock() + defer h.mu.Unlock() + for ch := range h.subs[key] { + select { + case ch <- v: + default: + } + } +} + +// Close ends every subscription to key. +func (h *Hub[T]) Close(key int64) { + h.mu.Lock() + defer h.mu.Unlock() + for ch := range h.subs[key] { + close(ch) + } + delete(h.subs, key) +} diff --git a/backend/internal/service/feed/feed_test.go b/backend/internal/service/feed/feed_test.go new file mode 100644 index 0000000..d8b33f2 --- /dev/null +++ b/backend/internal/service/feed/feed_test.go @@ -0,0 +1,53 @@ +package feed_test + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/agentrq/sql-mcp/backend/internal/service/feed" +) + +func TestPublishReachesOnlyTheKeysSubscribers(t *testing.T) { + h := feed.New[string]() + a, cancelA := h.Subscribe(1, 4) + b, cancelB := h.Subscribe(2, 4) + defer cancelA() + defer cancelB() + + h.Publish(1, "one") + h.Publish(3, "nobody") + + assert.Equal(t, "one", <-a) + assert.Empty(t, b) +} + +func TestASlowSubscriberLosesEvents(t *testing.T) { + h := feed.New[int]() + ch, cancel := h.Subscribe(1, 1) + defer cancel() + + h.Publish(1, 1) + h.Publish(1, 2) // dropped, the buffer is full + + assert.Equal(t, 1, <-ch) + assert.Empty(t, ch) +} + +func TestCancelAndCloseEndSubscriptions(t *testing.T) { + h := feed.New[int]() + a, cancelA := h.Subscribe(1, 1) + b, cancelB := h.Subscribe(1, 1) + + cancelA() + cancelA() // a second cancel is harmless + _, open := <-a + assert.False(t, open) + + h.Close(1) + _, open = <-b + assert.False(t, open) + cancelB() // after Close, cancel does not close it again + + h.Publish(1, 1) // no subscribers left +} diff --git a/backend/internal/testutil/testutil.go b/backend/internal/testutil/testutil.go index f85bc27..88118a1 100644 --- a/backend/internal/testutil/testutil.go +++ b/backend/internal/testutil/testutil.go @@ -165,6 +165,7 @@ type Env struct { Tokens *auth.TokenService Sealer *security.Sealer BaseURL string + DBPath string // the app database, a SQLite file } // Key is the fixed 32-byte encryption key tests use. @@ -178,7 +179,8 @@ func New(t *testing.T) *Env { t.Helper() var next atomic.Int64 next.Store(1_000_000) - repo, err := repository.Open("sqlite", filepath.Join(t.TempDir(), "app.db"), func() int64 { return next.Add(1) << 22 }) + dbPath := filepath.Join(t.TempDir(), "app.db") + repo, err := repository.Open("sqlite", dbPath, func() int64 { return next.Add(1) << 22 }) if err != nil { t.Fatal(err) } @@ -193,5 +195,5 @@ func New(t *testing.T) *Env { Tmphost: tmphost.New(th.URL, th.Client()), Config: crud.Config{BaseURL: baseURL, MaxRows: 100, InlineRows: 5, QueryTimeout: time.Second}, }) - return &Env{Ctrl: ctrl, Repo: repo, Driver: d, Tmphost: th, Tokens: tokens, Sealer: sealer, BaseURL: baseURL} + return &Env{Ctrl: ctrl, Repo: repo, Driver: d, Tmphost: th, Tokens: tokens, Sealer: sealer, BaseURL: baseURL, DBPath: dbPath} } diff --git a/docs/API.md b/docs/API.md index 7002a4b..7b2de9b 100644 --- a/docs/API.md +++ b/docs/API.md @@ -59,6 +59,15 @@ A database is one connection a person registered; it gets its own MCP server. | PUT | `/databases/:id/pii` | `{ "piiColumns": [PiiColumn] }` (replaces the whole set) | `{ "piiColumns": [PiiColumn] }` | | POST | `/databases/:id/query` | `{ "sql": "select ..." }` | `{ "columns": ["id"], "rows": [[1]], "rowCount": 1, "truncated": false }`; a query touching a PII column is 422 with the reason | | GET | `/databases/:id/queries?limit=50` | | `{ "queries": [{ "id", "source": "mcp" \| "console", "sql", "status": "ok" \| "rejected" \| "error", "reason", "rowCount", "resultUrl", "durationMs", "createdAt" }] }` | +| GET | `/databases/:id/queries/stream` | | `text/event-stream`: one `event: query` per query as it is recorded, `data` a query log entry as above and `id` its id; `: ping` comments every 25 s. 401 without a session, 404 for someone else's database | + +The query stream ends when the session's `at` cookie expires or the database +is deleted; reconnecting then is refused, so the browser renews the session +(`POST /auth/refresh`) and opens it again. A slow reader misses queries +rather than holding them up, and each (re)connect should reload +`/queries` to catch up. It is served by the standard mux, not Fiber (whose +adaptor buffers the whole response), and only by the process that ran the +query: run one replica, or the stream shows only that replica's queries. ## Settings diff --git a/frontend/src/api.js b/frontend/src/api.js index ba98fb9..6868c5d 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -183,6 +183,16 @@ export async function fetchQueries(id, limit = 50) { return data?.queries || [] } +/** The server-sent events stream of a database's queries, one `query` event each. */ +export function queryStreamUrl(id) { + return `${apiBase()}/databases/${enc(id)}/queries/stream` +} + +/** Swaps the refresh cookie for a new session; rejects when it cannot. */ +export function refreshSession() { + return request('/auth/refresh', { method: 'POST' }) +} + // --- Settings --------------------------------------------------------------- /** `{ connected, account?, serverUrl, connectedAt? }` */ diff --git a/frontend/src/components/ActivityPanel.vue b/frontend/src/components/ActivityPanel.vue index 90b040c..d190221 100644 --- a/frontend/src/components/ActivityPanel.vue +++ b/frontend/src/components/ActivityPanel.vue @@ -1,27 +1,46 @@ diff --git a/frontend/src/components/QueryDetail.vue b/frontend/src/components/QueryDetail.vue new file mode 100644 index 0000000..de3251b --- /dev/null +++ b/frontend/src/components/QueryDetail.vue @@ -0,0 +1,75 @@ + + + + diff --git a/frontend/src/components/queryStatus.js b/frontend/src/components/queryStatus.js new file mode 100644 index 0000000..72659cf --- /dev/null +++ b/frontend/src/components/queryStatus.js @@ -0,0 +1,6 @@ +// Badge colours of a logged query's status, shared by the list and its detail. +export const STATUS_STYLES = { + ok: 'bg-green-50 text-green-700 dark:bg-green-500/10 dark:text-green-400', + rejected: 'bg-amber-50 text-amber-700 dark:bg-amber-500/10 dark:text-amber-400', + error: 'bg-red-50 text-red-700 dark:bg-red-500/10 dark:text-red-400', +} diff --git a/frontend/src/composables/useFormat.js b/frontend/src/composables/useFormat.js index 397e692..4ccefd9 100644 --- a/frontend/src/composables/useFormat.js +++ b/frontend/src/composables/useFormat.js @@ -16,6 +16,14 @@ export function formatDateTime(value) { return d.toLocaleString(undefined, { year: 'numeric', month: 'short', day: 'numeric', hour: '2-digit', minute: '2-digit' }) } +/** A timestamp in full, to the second; '' for nothing, the raw value when unparseable. */ +export function formatFullDateTime(value) { + if (!value) return '' + const d = new Date(value) + if (Number.isNaN(d.getTime())) return String(value) + return d.toLocaleString(undefined, { year: 'numeric', month: 'short', day: 'numeric', hour: '2-digit', minute: '2-digit', second: '2-digit' }) +} + /** Milliseconds as "12 ms" or "1.4 s". */ export function formatDuration(ms) { if (ms === null || ms === undefined || Number.isNaN(Number(ms))) return '' diff --git a/frontend/src/composables/useQueryStream.js b/frontend/src/composables/useQueryStream.js new file mode 100644 index 0000000..7d7144d --- /dev/null +++ b/frontend/src/composables/useQueryStream.js @@ -0,0 +1,105 @@ +/** + * A database's queries as they happen, over server-sent events. + * + * The browser retries a dropped stream by itself. What it cannot do is renew + * the session: when the server refuses the stream (an expired cookie is a + * 401), EventSource gives up for good. So a refused stream renews the session + * once and reconnects, backing off, and after `maxFailures` refusals in a row + * the stream is `offline` until `start()` again. + */ +import { ref } from 'vue' + +/** + * @param {object} deps + * @param {() => string} deps.url the stream's URL + * @param {(query: object) => void} deps.onQuery each query as it is recorded + * @param {() => void} [deps.onOpen] the stream is (re)connected: catch up on what was missed + * @param {() => Promise} [deps.refresh] renews the session + * @param {typeof EventSource} [deps.EventSourceImpl] + * @param {number} [deps.maxFailures] + * @param {number} [deps.backoffMs] the wait before the nth retry is n times this + */ +export function createQueryStream({ + url, + onQuery, + onOpen = () => {}, + refresh = async () => {}, + EventSourceImpl = globalThis.EventSource, + maxFailures = 3, + backoffMs = 1000, +}) { + /** 'connecting' | 'live' | 'offline' */ + const status = ref('connecting') + let source = null + let timer = null + let failures = 0 + let stopped = false + + function close() { + clearTimeout(timer) + timer = null + if (source) source.close() + source = null + } + + function connect() { + close() + if (!EventSourceImpl) { + status.value = 'offline' + return + } + status.value = 'connecting' + const es = new EventSourceImpl(url()) + source = es + es.onopen = () => { + failures = 0 + status.value = 'live' + onOpen() + } + es.addEventListener('query', (e) => { + let query + try { + query = JSON.parse(e.data) + } catch { + return + } + onQuery(query) + }) + es.onerror = () => { + if (es !== source) return + // Still CONNECTING: the browser is retrying a dropped stream itself. + if (es.readyState !== EventSourceImpl.CLOSED) { + status.value = 'connecting' + return + } + close() + failures += 1 + if (failures > maxFailures) { + status.value = 'offline' + return + } + status.value = 'connecting' + const wait = failures * backoffMs + Promise.resolve() + .then(refresh) + .catch(() => {}) + .then(() => { + if (!stopped) timer = setTimeout(connect, wait) + }) + } + } + + return { + status, + /** Connects, or tries again after the stream went offline. */ + start() { + stopped = false + failures = 0 + connect() + }, + stop() { + stopped = true + close() + }, + } +} diff --git a/frontend/test/ActivityPanel.test.js b/frontend/test/ActivityPanel.test.js new file mode 100644 index 0000000..03e5f22 --- /dev/null +++ b/frontend/test/ActivityPanel.test.js @@ -0,0 +1,138 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { mount } from '@vue/test-utils' +import ActivityPanel from '../src/components/ActivityPanel.vue' +import { useToasts } from '../src/composables/useToasts' +import { flush, mockFetch } from './helpers' + +class FakeEventSource { + static CLOSED = 2 + static instances = [] + constructor(url) { + this.url = url + this.listeners = {} + FakeEventSource.instances.push(this) + } + addEventListener(type, fn) { this.listeners[type] = fn } + close() { this.closed = true } +} + +const q = (id, createdAt, extra = {}) => ({ + id, createdAt, source: 'mcp', sql: `select ${id} from orders`, status: 'ok', reason: '', rowCount: 3, durationMs: 12, resultUrl: '', ...extra, +}) + +const stream = () => FakeEventSource.instances.at(-1) + +describe('ActivityPanel', () => { + beforeEach(() => { + FakeEventSource.instances = [] + vi.stubGlobal('EventSource', FakeEventSource) + }) + afterEach(() => { + vi.unstubAllGlobals() + useToasts().toasts.value = [] + }) + + async function mountPanel(routes) { + const f = mockFetch(routes) + const w = mount(ActivityPanel, { props: { databaseId: 'd1' }, attachTo: document.body }) + await flush(); await flush() + return { w, f } + } + + it('adds queries as they arrive on the stream', async () => { + const { w } = await mountPanel({ 'GET /databases/d1/queries?limit=50': { queries: [q('a', '2026-10-23T04:00:00Z')] } }) + expect(stream().url).toBe('/api/v1/databases/d1/queries/stream') + expect(w.find('[data-test="stream-status"]').attributes('data-status')).toBe('connecting') + + stream().onopen() + await flush(); await flush() + expect(w.find('[data-test="stream-status"]').text()).toBe('Live') + + stream().listeners.query({ data: JSON.stringify(q('b', '2026-10-23T04:01:00Z')) }) + stream().listeners.query({ data: JSON.stringify(q('b', '2026-10-23T04:01:00Z')) }) // a repeat is ignored + await flush() + const rows = w.findAll('[data-test="query-row"]') + expect(rows).toHaveLength(2) + expect(rows[0].text()).toContain('select b from orders') + w.unmount() + expect(stream().closed).toBe(true) + }) + + it('keeps a streamed query the catch-up load missed', async () => { + let resolve + const { w } = await mountPanel({ 'GET /databases/d1/queries?limit=50': () => new Promise((r) => { resolve = r }) }) + stream().listeners.query({ data: JSON.stringify(q('new', '2026-10-23T04:05:00Z')) }) + resolve({ queries: [q('old', '2026-10-23T04:00:00Z')] }) + await flush(); await flush() + expect(w.findAll('[data-test="query-row"]').map((r) => r.text().includes('select new'))).toEqual([true, false]) + }) + + it('shows the whole query in the sidebar when a row is clicked', async () => { + const long = `select ${'col, '.repeat(200)}id from orders where note = 'x'` + const { w } = await mountPanel({ + 'GET /databases/d1/queries?limit=50': { queries: [ + q('a', '2026-10-23T04:00:00Z', { sql: long, resultUrl: 'https://tmphost.io/r/1' }), + q('b', '2026-10-23T03:00:00Z', { status: 'rejected', reason: 'query reads PII column public.users.email', sql: 'select email from users' }), + ] }, + }) + expect(w.find('[data-test="query-detail"]').exists()).toBe(false) + + await w.findAll('[data-test="query-row"]')[0].trigger('click') + expect(w.find('[data-test="detail-sql"]').text()).toBe(long) + expect(w.find('[data-test="detail-rows"]').text()).toBe('3') + expect(w.find('[data-test="detail-result"]').attributes('href')).toBe('https://tmphost.io/r/1') + expect(w.findAll('[data-test="query-row"]')[0].attributes('aria-selected')).toBe('true') + + const writeText = vi.fn(async () => {}) + vi.stubGlobal('navigator', { clipboard: { writeText } }) + await w.find('[data-test="copy-sql"]').trigger('click') + expect(writeText).toHaveBeenCalledWith(long) + + await w.findAll('[data-test="query-row"]')[1].trigger('keydown', { key: 'Enter' }) + expect(w.find('[data-test="detail-sql"]').text()).toBe('select email from users') + expect(w.find('[data-test="detail-reason"]').text()).toContain('public.users.email') + expect(w.find('[data-test="detail-status-rejected"]').exists()).toBe(true) + expect(w.find('[data-test="detail-rows"]').exists()).toBe(false) + + window.dispatchEvent(new KeyboardEvent('keydown', { key: 'a' })) + await flush() + expect(w.find('[data-test="query-detail"]').exists()).toBe(true) + window.dispatchEvent(new KeyboardEvent('keydown', { key: 'Escape' })) + await flush() + expect(w.find('[data-test="query-detail"]').exists()).toBe(false) + + await w.findAll('[data-test="query-row"]')[0].trigger('click') + await w.find('[data-test="close-detail"]').trigger('click') + expect(w.find('[data-test="query-detail"]').exists()).toBe(false) + await w.findAll('[data-test="query-row"]')[0].trigger('click') + await w.findAll('[data-test="query-row"]')[0].trigger('click') + expect(w.find('[data-test="query-detail"]').exists()).toBe(false) // a second click closes it + w.unmount() + }) + + it('offers to reconnect once the stream gives up', async () => { + vi.useFakeTimers() + try { + mockFetch({ 'GET /databases/d1/queries?limit=50': { queries: [] }, 'POST /auth/refresh': [401, {}] }) + const w = mount(ActivityPanel, { props: { databaseId: 'd1' } }) + for (let i = 0; i < 4; i++) { + stream().readyState = FakeEventSource.CLOSED + stream().onerror() + await vi.advanceTimersByTimeAsync(5000) + } + expect(w.find('[data-test="stream-status"]').text()).toBe('Offline') + expect(w.find('[data-test="empty"]').exists()).toBe(true) + const before = FakeEventSource.instances.length + await w.find('[data-test="reconnect"]').trigger('click') + expect(FakeEventSource.instances).toHaveLength(before + 1) + w.unmount() + } finally { + vi.useRealTimers() + } + }) + + it('shows a failed load', async () => { + const { w } = await mountPanel({ 'GET /databases/d1/queries?limit=50': [500, { error: { message: 'boom' } }] }) + expect(w.text()).toContain('boom') + }) +}) diff --git a/frontend/test/api.test.js b/frontend/test/api.test.js index 36311e1..c761525 100644 --- a/frontend/test/api.test.js +++ b/frontend/test/api.test.js @@ -128,6 +128,15 @@ describe('api', () => { expect(await api.fetchQueries('a')).toEqual([]) }) + it('points the query stream at the database and renews the session', async () => { + expect(api.queryStreamUrl('a/b')).toBe('/api/v1/databases/a%2Fb/queries/stream') + const f = mockFetch({ 'POST /auth/refresh': { status: 'ok' } }) + await api.refreshSession() + expect(f.mock.calls[0][1].method).toBe('POST') + mockFetch({}) + await expect(api.refreshSession()).rejects.toBeInstanceOf(api.ApiError) + }) + it('escapes ids in paths', async () => { const f = mockFetch({}) await api.fetchDatabase('a/b').catch(() => {}) diff --git a/frontend/test/composables.test.js b/frontend/test/composables.test.js index 0ab678d..5f9311c 100644 --- a/frontend/test/composables.test.js +++ b/frontend/test/composables.test.js @@ -1,7 +1,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { useToasts } from '../src/composables/useToasts' import { copyToClipboard } from '../src/composables/useClipboard' -import { formatCell, formatDateTime, formatDuration, toKebabCase } from '../src/composables/useFormat' +import { formatCell, formatDateTime, formatDuration, formatFullDateTime, toKebabCase } from '../src/composables/useFormat' import { useQueryConsole } from '../src/composables/useQueryConsole' import { ApiError } from '../src/api' @@ -58,6 +58,9 @@ describe('useFormat', () => { expect(formatDateTime('')).toBe('') expect(formatDateTime('nope')).toBe('nope') expect(formatDateTime('2026-01-02T03:04:05Z')).toMatch(/2026/) + expect(formatFullDateTime('')).toBe('') + expect(formatFullDateTime('nope')).toBe('nope') + expect(formatFullDateTime('2026-01-02T03:04:05Z')).toMatch(/05/) expect(formatDuration(null)).toBe('') expect(formatDuration('x')).toBe('') expect(formatDuration(12.4)).toBe('12 ms') diff --git a/frontend/test/useQueryStream.test.js b/frontend/test/useQueryStream.test.js new file mode 100644 index 0000000..91dec59 --- /dev/null +++ b/frontend/test/useQueryStream.test.js @@ -0,0 +1,159 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createQueryStream } from '../src/composables/useQueryStream' + +/** A stand-in EventSource the test drives by hand. */ +class FakeEventSource { + static CONNECTING = 0 + static OPEN = 1 + static CLOSED = 2 + static instances = [] + + constructor(url) { + this.url = url + this.readyState = FakeEventSource.CONNECTING + this.listeners = {} + this.closed = false + FakeEventSource.instances.push(this) + } + + addEventListener(type, fn) { + this.listeners[type] = fn + } + + close() { + this.closed = true + this.readyState = FakeEventSource.CLOSED + } + + open() { + this.readyState = FakeEventSource.OPEN + this.onopen() + } + + emit(type, data) { + this.listeners[type]({ data }) + } + + /** The server refused it: EventSource gives up. */ + refuse() { + this.readyState = FakeEventSource.CLOSED + this.onerror() + } + + /** The stream dropped: the browser retries by itself. */ + drop() { + this.readyState = FakeEventSource.CONNECTING + this.onerror() + } +} + +const latest = () => FakeEventSource.instances.at(-1) + +function setup(overrides = {}) { + const onQuery = vi.fn() + const onOpen = vi.fn() + const refresh = vi.fn(async () => {}) + const stream = createQueryStream({ + url: () => '/api/v1/databases/d1/queries/stream', + onQuery, + onOpen, + refresh, + EventSourceImpl: FakeEventSource, + maxFailures: 2, + backoffMs: 100, + ...overrides, + }) + return { stream, onQuery, onOpen, refresh } +} + +describe('createQueryStream', () => { + beforeEach(() => { + FakeEventSource.instances = [] + vi.useFakeTimers() + }) + afterEach(() => vi.useRealTimers()) + + it('delivers queries once the stream is open', () => { + const { stream, onQuery, onOpen } = setup() + stream.start() + expect(latest().url).toBe('/api/v1/databases/d1/queries/stream') + expect(stream.status.value).toBe('connecting') + latest().open() + expect(stream.status.value).toBe('live') + expect(onOpen).toHaveBeenCalledTimes(1) + latest().emit('query', '{"id":"q1"}') + latest().emit('query', 'not json') + expect(onQuery).toHaveBeenCalledTimes(1) + expect(onQuery).toHaveBeenCalledWith({ id: 'q1' }) + }) + + it('lets the browser retry a dropped stream', () => { + const { stream } = setup() + stream.start() + latest().open() + latest().drop() + expect(stream.status.value).toBe('connecting') + expect(FakeEventSource.instances).toHaveLength(1) + }) + + it('renews the session and reconnects after a refusal, then gives up', async () => { + const { stream, refresh, onOpen } = setup() + stream.start() + latest().refuse() + expect(latest().closed).toBe(true) + await vi.advanceTimersByTimeAsync(100) + expect(refresh).toHaveBeenCalledTimes(1) + expect(FakeEventSource.instances).toHaveLength(2) + + latest().refuse() + await vi.advanceTimersByTimeAsync(200) + expect(FakeEventSource.instances).toHaveLength(3) + latest().refuse() + await vi.advanceTimersByTimeAsync(1000) + expect(stream.status.value).toBe('offline') + expect(FakeEventSource.instances).toHaveLength(3) + + stream.start() + latest().open() + expect(stream.status.value).toBe('live') + expect(onOpen).toHaveBeenCalledTimes(1) + }) + + it('reconnects even when the session cannot be renewed', async () => { + const { stream } = setup({ refresh: vi.fn(async () => { throw new Error('401') }) }) + stream.start() + latest().refuse() + await vi.advanceTimersByTimeAsync(100) + expect(FakeEventSource.instances).toHaveLength(2) + }) + + it('stays down once stopped', async () => { + const { stream } = setup() + stream.start() + const first = latest() + first.refuse() + stream.stop() + await vi.advanceTimersByTimeAsync(1000) + expect(FakeEventSource.instances).toHaveLength(1) + first.onerror() // a late error from a stream already replaced is ignored + }) + + it('is offline without EventSource, and works with the defaults', () => { + const { stream } = setup({ EventSourceImpl: undefined }) + stream.start() + expect(stream.status.value).toBe('offline') + + vi.stubGlobal('EventSource', undefined) + const bare = createQueryStream({ url: () => '/x', onQuery: () => {} }) + bare.start() + expect(bare.status.value).toBe('offline') + vi.unstubAllGlobals() + + const defaults = createQueryStream({ url: () => '/x', onQuery: () => {}, EventSourceImpl: FakeEventSource }) + defaults.start() + latest().open() + latest().refuse() + expect(defaults.status.value).toBe('connecting') + defaults.stop() + }) +})