From c62900e6132c56c8bc229087674a3fb91deb3427 Mon Sep 17 00:00:00 2001 From: Raphael Westphal Date: Tue, 18 Aug 2026 13:42:20 +0200 Subject: [PATCH] daemon: golang WS hub, REST, gitlab, docker spawner, sqlite (18/18 tests) --- .gitignore | 1 + daemon/Dockerfile | 22 ++ daemon/README.md | 100 ++++++ daemon/api.go | 366 ++++++++++++++++++++++ daemon/api_test.go | 135 +++++++++ daemon/docker.go | 539 +++++++++++++++++++++++++++++++++ daemon/gitlab.go | 184 +++++++++++ daemon/gitlab_test.go | 123 ++++++++ daemon/go.mod | 47 +++ daemon/go.sum | 147 +++++++++ daemon/hub.go | 621 ++++++++++++++++++++++++++++++++++++++ daemon/hub_test.go | 332 ++++++++++++++++++++ daemon/main.go | 86 ++++++ daemon/store.go | 239 +++++++++++++++ daemon/store_test.go | 134 ++++++++ daemon/webdist/.gitkeep | 0 daemon/webdist/index.html | 24 ++ 17 files changed, 3100 insertions(+) create mode 100644 daemon/Dockerfile create mode 100644 daemon/README.md create mode 100644 daemon/api.go create mode 100644 daemon/api_test.go create mode 100644 daemon/docker.go create mode 100644 daemon/gitlab.go create mode 100644 daemon/gitlab_test.go create mode 100644 daemon/go.mod create mode 100644 daemon/go.sum create mode 100644 daemon/hub.go create mode 100644 daemon/hub_test.go create mode 100644 daemon/main.go create mode 100644 daemon/store.go create mode 100644 daemon/store_test.go create mode 100644 daemon/webdist/.gitkeep create mode 100644 daemon/webdist/index.html diff --git a/.gitignore b/.gitignore index 727ec79..9f00bf1 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ node_modules dist *.db .pi/ +daemon/.pi-scratch/ diff --git a/daemon/Dockerfile b/daemon/Dockerfile new file mode 100644 index 0000000..c147f63 --- /dev/null +++ b/daemon/Dockerfile @@ -0,0 +1,22 @@ +# lvmh daemon image. Build from repo root: +# docker build -f daemon/Dockerfile -t lvmh-daemon:latest . +# To bake the real web UI, copy web/dist over the placeholder first: +# cp -r web/dist daemon/webdist/ +FROM golang:1.24-alpine AS build +WORKDIR /src +COPY daemon/go.mod daemon/go.sum ./ +RUN go mod download +COPY daemon/ ./ +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/lvmh-daemon . + +FROM alpine:3.21 +RUN apk add --no-cache ca-certificates git +WORKDIR /app +COPY --from=build /out/lvmh-daemon /app/lvmh-daemon +# Placeholder UI; replace via build (cp web/dist daemon/webdist) or mount at +# /app/web-dist and rely on the --webdist default. +COPY --from=build /src/webdist /app/web-dist +VOLUME /data +EXPOSE 8686 +ENV LVMH_DB=/data/lvmh.db +ENTRYPOINT ["/app/lvmh-daemon"] diff --git a/daemon/README.md b/daemon/README.md new file mode 100644 index 0000000..96bf2a2 --- /dev/null +++ b/daemon/README.md @@ -0,0 +1,100 @@ +# lvmh daemon + +Go daemon binding the lvmh wire protocol (`../PROTOCOL.md`): agent websocket +(`/agent/ws`), REST API (`/api/*`), web websocket (`/ws`), GitLab PAT +management and worker-container spawning over the Docker API. + +## Layout + +| file | role | +| --- | --- | +| `main.go` | flags, env wiring, HTTP server lifecycle | +| `hub.go` | agent WS hub (`/agent/ws`) + web WS hub (`/ws`), prompt/abort routing, online state | +| `api.go` | REST handlers, bearer auth, recover middleware, embedded web UI | +| `gitlab.go` | PAT connect/validate/list member projects | +| `docker.go` | spawn pipeline: clone → ensure image → create/start container, job status | +| `store.go` | SQLite (modernc, CGO-free): sessions, events, settings, containers | +| `webdist/` | embedded web UI placeholder (replace with `web/dist` at build time) | + +## Build & run + +```sh +cd daemon +go build ./... + +LVMH_TOKEN=secret LVMH_DB=/tmp/lvmh.db ./lvmh-daemon --addr :8686 +``` + +Flags: `--addr` (default `:8686`), `--db` (default `$LVMH_DB` or `/data/lvmh.db`), +`--webdist` (default `/app/web-dist` when that directory exists, else the +embedded `webdist/`). + +Docker image (multi-stage, `CGO_ENABLED=0`): + +```sh +docker build -f daemon/Dockerfile -t lvmh-daemon:latest . +``` + +To bake the real web UI into the binary, copy the built frontend over the +placeholder before building: `cp -r web/dist daemon/webdist/`. Otherwise mount +it at `/app/web-dist` (the default `--webdist` path) or pass `--webdist`. + +## Environment + +| var | default | meaning | +| --- | --- | --- | +| `LVMH_TOKEN` | — (required) | bearer token for REST + WS auth | +| `LVMH_DB` | `/data/lvmh.db` | sqlite path (WAL mode) | +| `GITLAB_BASE_URL` | `https://git.westphal.fr` | GitLab instance | +| `LVMH_WORKER_DOCKERFILE` | `/app/build/docker/worker.Dockerfile` | image build recipe | +| `LVMH_WORKER_CONTEXT` | parent of dockerfile dir | docker build context | +| `LVMH_REPO_DIR` | `/data/repos` | host-side clone cache (`` subdirs) | +| `LVMH_CONTAINER_LVMH_URL` | `ws://lvmh:8686/agent/ws` | daemon URL handed to spawned containers | +| `LVMH_CONTAINER_NETWORK` | `lvmh-net` | network for spawned containers | +| `LVMH_WORKER_MODELS` | `/app/build/docker/worker-models.json` | mounted read-only as `models.json` when present | +| `ZAI_RENAUD_API_KEY` | — | injected into spawned containers | +| `DOCKER_HOST` | — | honored by the docker client | + +## Spawn pipeline + +`POST /api/spawn {repo, branch?}` returns `{sessionId, containerId:""}` +immediately (async). The job walks `cloning → building → creating → running` +(or `error` + message) in `GET /api/spawn/status` and on the web WS. + +1. Per-repo-slug mutex; `git clone`/`git pull --ff-only` into `$LVMH_REPO_DIR/`. +2. Ensure image `lvmh-worker:latest` — built from `LVMH_WORKER_DOCKERFILE` + when missing (502-style clear error from `POST /api/spawn` if neither image + nor Dockerfile exist). +3. Named volume `lvmh-repo-` (seeded from the clone on first use), + `lvmh-sessions` volume, `lvmh.session=` label, network + `lvmh-net`, env `ZAI_RENAUD_API_KEY`, `LVMH_TOKEN`, `LVMH_URL`, + `LVMH_SESSION_ID`. +4. `DELETE /api/sessions/:id/container` stops and removes the container. + +## Tests + +```sh +go vet ./... && staticcheck ./... && go test ./... -count=1 +``` + +Covered: store persistence + replay-after-seq, agent handshake +(hello→welcome.lastSeq), reconnect replacement, live `message_update` fan-out +(no persistence), prompt routing + 409 offline, bearer 401s, GitLab user/ +projects mapping against an httptest upstream. + +## Protocol ambiguities (daemon-side resolutions) + +- **Envelope payload shape**: payload fields are flattened into the envelope + object (as the `hello` example shows). The daemon persists the full frame + verbatim; `GET /api/sessions/:id/events` replays it as-is. +- **`events` frame `after`**: exclusive lower bound = `seq` of the first event + in the batch minus 1, matching the REST `?after=` semantics. +- **`POST /api/spawn` response**: returns immediately with `{sessionId, + containerId: ""}` — `containerId` appears later via `/api/spawn/status` + (async spawn per spec; the protocol table's synchronous shape is not + achievable for clones that take minutes). +- **Events persistence**: everything except `message_update`; both live and + replayed `message_update` content is reconstructed from `message_end`. +- **Repo workspace**: clone cache on the daemon filesystem under + `LVMH_REPO_DIR`; the container mounts named volume `lvmh-repo-` seeded + from that clone on first spawn (protocol: "clone into volume"). diff --git a/daemon/api.go b/daemon/api.go new file mode 100644 index 0000000..b52e255 --- /dev/null +++ b/daemon/api.go @@ -0,0 +1,366 @@ +package main + +// api.go — REST /api/* handlers, bearer auth, static web serving. + +import ( + "crypto/subtle" + "embed" + "encoding/json" + "errors" + "fmt" + "io/fs" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" +) + +// Route names, event query defaults and limits. +const ( + envToken string = "LVMH_TOKEN" + + defaultEventsAfter int64 = 0 + defaultEventsLimit int = 1000 + maxEventsLimit int = 10000 + maxBodyBytes int64 = 1 << 20 + webIndexFallback string = "index.html" +) + +//go:embed webdist +var embeddedWeb embed.FS + +// Server wires store, hub, spawner and gitlab client into HTTP handlers. +type Server struct { + store *Store + hub *Hub + spawn *Spawner + gitlab *GitLab +} + +// NewServer builds the Server and cross-links hub callbacks. +func NewServer(store *Store, hub *Hub, spawn *Spawner, gl *GitLab) *Server { + s := &Server{store: store, hub: hub, spawn: spawn, gitlab: gl} + hub.SpawnStatus = spawn.JobsSnapshot + return s +} + +// Routes assembles the full mux with auth middleware on /api and /agent. +func (s *Server) Routes(webdist string) http.Handler { + mux := http.NewServeMux() + + mux.HandleFunc("GET /agent/ws", s.hub.ServeAgentWS) + mux.HandleFunc("GET /ws", s.handleWebWS) + + api := http.NewServeMux() + api.HandleFunc("GET /api/sessions", s.handleSessions) + api.HandleFunc("GET /api/sessions/{id}/events", s.handleSessionEvents) + api.HandleFunc("POST /api/sessions/{id}/prompt", s.handlePrompt) + api.HandleFunc("POST /api/sessions/{id}/abort", s.handleAbort) + api.HandleFunc("DELETE /api/sessions/{id}/container", s.handleDeleteContainer) + api.HandleFunc("POST /api/spawn", s.handleSpawn) + api.HandleFunc("GET /api/spawn/status", s.handleSpawnStatus) + api.HandleFunc("GET /api/gitlab/status", s.handleGitLabStatus) + api.HandleFunc("POST /api/gitlab/connect", s.handleGitLabConnect) + api.HandleFunc("DELETE /api/gitlab/connect", s.handleGitLabDisconnect) + api.HandleFunc("GET /api/gitlab/repos", s.handleGitLabRepos) + mux.Handle("/api/", s.bearerAuth(api)) + + mux.Handle("/", s.webHandler(webdist)) + return recoverMiddleware(mux) +} + +// recoverMiddleware converts panics into 500s instead of killing the server. +func recoverMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer func() { + if rec := recover(); rec != nil { + writeError(w, http.StatusInternalServerError, fmt.Sprintf("internal error: %v", rec)) + } + }() + next.ServeHTTP(w, r) + }) +} + +// bearerAuth rejects non-matching bearer tokens with 401. +func (s *Server) bearerAuth(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !tokenMatches(r.Header.Get("Authorization")) { + writeError(w, http.StatusUnauthorized, "unauthorized") + return + } + next.ServeHTTP(w, r) + }) +} + +// TokenMatches validates an Authorization header value against LVMH_TOKEN. +func tokenMatches(header string) bool { + const prefix = "Bearer " + if !strings.HasPrefix(header, prefix) { + return false + } + token := strings.TrimSpace(strings.TrimPrefix(header, prefix)) + want := strings.TrimSpace(daemonToken) + if want == "" || token == "" { + return false + } + return subtle.ConstantTimeCompare([]byte(token), []byte(want)) == 1 +} + +// handleWebWS upgrades /ws with token from query param (?token=). +func (s *Server) handleWebWS(w http.ResponseWriter, r *http.Request) { + token := r.URL.Query().Get("token") + want := strings.TrimSpace(daemonToken) + if want == "" || token == "" || + subtle.ConstantTimeCompare([]byte(token), []byte(want)) != 1 { + // Reject before upgrade: browsers can't read WS close codes well. + writeError(w, http.StatusUnauthorized, "unauthorized") + return + } + s.hub.ServeWebWS(w, r) +} + +func writeError(w http.ResponseWriter, status int, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]string{"error": msg}) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +func decodeBody(w http.ResponseWriter, r *http.Request, dst any) bool { + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + dec := json.NewDecoder(r.Body) + if err := dec.Decode(dst); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body: "+err.Error()) + return false + } + return true +} + +func (s *Server) handleSessions(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.hub.SessionsView()) +} + +func (s *Server) handleSessionEvents(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + after := defaultEventsAfter + if raw := r.URL.Query().Get("after"); raw != "" { + v, err := strconv.ParseInt(raw, 10, 64) + if err != nil || v < 0 { + writeError(w, http.StatusBadRequest, "invalid after") + return + } + after = v + } + limit := defaultEventsLimit + if raw := r.URL.Query().Get("limit"); raw != "" { + v, err := strconv.Atoi(raw) + if err != nil || v <= 0 { + writeError(w, http.StatusBadRequest, "invalid limit") + return + } + limit = v + } + if limit > maxEventsLimit { + limit = maxEventsLimit + } + events, err := s.store.EventsAfter(id, after, limit) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if events == nil { + events = []Event{} + } + out := make([]map[string]any, 0, len(events)) + for _, e := range events { + var payload map[string]any + if len(e.Payload) > 0 { + if err := json.Unmarshal(e.Payload, &payload); err != nil { + payload = map[string]any{} + } + } + frameMap := map[string]any{} + for k, v := range payload { + frameMap[k] = v + } + frameMap["v"] = daemonVersion + frameMap["sessionId"] = e.SessionID + frameMap["seq"] = e.Seq + frameMap["ts"] = e.TS + frameMap["type"] = e.Type + out = append(out, frameMap) + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handlePrompt(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + var body struct { + Message string `json:"message"` + } + if !decodeBody(w, r, &body) { + return + } + if strings.TrimSpace(body.Message) == "" { + writeError(w, http.StatusBadRequest, "message required") + return + } + if err := s.hub.Prompt(id, body.Message); err != nil { + if errors.Is(err, ErrOffline) { + writeError(w, http.StatusConflict, "session offline") + return + } + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +func (s *Server) handleAbort(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if err := s.hub.Abort(id); err != nil { + if errors.Is(err, ErrOffline) { + writeError(w, http.StatusConflict, "session offline") + return + } + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +func (s *Server) handleDeleteContainer(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if err := s.spawn.RemoveSession(r.Context(), id); err != nil { + if errors.Is(err, errNoContainer) { + writeError(w, http.StatusNotFound, "no container for session") + return + } + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +func (s *Server) handleSpawn(w http.ResponseWriter, r *http.Request) { + var body struct { + Repo string `json:"repo"` + Branch string `json:"branch"` + } + if !decodeBody(w, r, &body) { + return + } + if !validRepoPath(body.Repo) { + writeError(w, http.StatusBadRequest, "repo must look like group/project") + return + } + res, err := s.spawn.Start(r.Context(), body.Repo, body.Branch) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusCreated, res) +} + +func (s *Server) handleSpawnStatus(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.spawn.JobsSnapshot()) +} + +func (s *Server) handleGitLabStatus(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.gitlab.Status()) +} + +func (s *Server) handleGitLabConnect(w http.ResponseWriter, r *http.Request) { + var body struct { + Token string `json:"token"` + } + if !decodeBody(w, r, &body) { + return + } + if strings.TrimSpace(body.Token) == "" { + writeError(w, http.StatusBadRequest, "token required") + return + } + username, err := s.gitlab.Connect(r.Context(), body.Token) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]string{"username": username}) +} + +func (s *Server) handleGitLabDisconnect(w http.ResponseWriter, r *http.Request) { + if err := s.gitlab.Disconnect(); err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) +} + +func (s *Server) handleGitLabRepos(w http.ResponseWriter, r *http.Request) { + repos, err := s.gitlab.Repos(r.Context()) + if err != nil { + status := http.StatusInternalServerError + var httpErr *GitLabError + if errors.As(err, &httpErr) { + status = http.StatusBadGateway + } + if errors.Is(err, errNotConnected) { + status = http.StatusConflict + } + writeError(w, status, err.Error()) + return + } + if repos == nil { + repos = []GitLabRepo{} + } + writeJSON(w, http.StatusOK, repos) +} + +// webHandler serves the embedded web/dist, or an override directory when +// webdist is non-empty (dev mode / mounted volume). +func (s *Server) webHandler(webdist string) http.Handler { + if webdist != "" { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Guard against path traversal outside the override dir. + cleaned := filepath.Clean("/" + r.URL.Path) + full := filepath.Join(webdist, cleaned) + if !strings.HasPrefix(full, filepath.Clean(webdist)) { + http.NotFound(w, r) + return + } + st, err := os.Stat(full) + if err == nil && !st.IsDir() { + http.ServeFile(w, r, full) + return + } + http.ServeFile(w, r, filepath.Join(webdist, webIndexFallback)) + }) + } + sub, err := fs.Sub(embeddedWeb, "webdist") + if err != nil { + panic(err) + } + fileServer := http.FileServer(http.FS(sub)) + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + path := strings.TrimPrefix(r.URL.Path, "/") + if path == "" { + path = webIndexFallback + } + if _, err := fs.Stat(sub, path); err != nil { + r.URL.Path = "/" + webIndexFallback // SPA fallback + } + fileServer.ServeHTTP(w, r) + }) +} + +var _ = daemonVersion + +// daemonToken is set by main from LVMH_TOKEN (single process, read-only). +var daemonToken string diff --git a/daemon/api_test.go b/daemon/api_test.go new file mode 100644 index 0000000..95c9ed4 --- /dev/null +++ b/daemon/api_test.go @@ -0,0 +1,135 @@ +package main + +// api_test.go — bearer auth 401s, events endpoint, spawn validation. + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestAPIAuthRejectsMissingOrBadToken(t *testing.T) { + ts, _ := newTestServer(t) + + cases := []struct { + name string + header string + }{ + {"none", ""}, + {"wrong", "Bearer nope"}, + {"not-bearer", testToken}, + {"empty", "Bearer "}, + } + for _, tc := range cases { + req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/sessions", nil) + if tc.header != "" { + req.Header.Set("Authorization", tc.header) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("%s: status = %d, want 401", tc.name, resp.StatusCode) + } + } +} + +func TestAPIEventsEndpointParams(t *testing.T) { + ts, _ := newTestServer(t) + + get := func(path string) int { + req, _ := http.NewRequest(http.MethodGet, ts.URL+path, nil) + req.Header.Set("Authorization", "Bearer "+testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("get %s: %v", path, err) + } + resp.Body.Close() + return resp.StatusCode + } + if code := get("/api/sessions/s1/events?after=abc"); code != http.StatusBadRequest { + t.Fatalf("after=abc → %d, want 400", code) + } + if code := get("/api/sessions/s1/events?limit=-1"); code != http.StatusBadRequest { + t.Fatalf("limit=-1 → %d, want 400", code) + } + if code := get("/api/sessions/unknown/events"); code != http.StatusOK { + t.Fatalf("unknown session → %d, want 200 with []", code) + } +} + +func TestAPIEventsReplayShape(t *testing.T) { + ts, store := newTestServer(t) + + // Simulate a connected session by writing directly through the store. + name := "sess" + if err := store.UpsertSession(SessionInfo{ID: "s1", Name: &name, Cwd: "/w", Model: "glm-5.3", Provider: "zai-renaud", StartedAt: 1}); err != nil { + t.Fatalf("upsert: %v", err) + } + if err := store.AppendEvent(Event{SessionID: "s1", Seq: 1, TS: 10, Type: evMessageEnd, Payload: []byte(`{"message":{"role":"user","id":"m1","text":"hi"}}`)}); err != nil { + t.Fatalf("append: %v", err) + } + + req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/sessions/s1/events?after=0&limit=10", nil) + req.Header.Set("Authorization", "Bearer "+testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("events: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("events status = %d", resp.StatusCode) + } + if ct := resp.Header.Get("Content-Type"); ct != "application/json" { + t.Fatalf("content-type = %q", ct) + } + var frames []map[string]any + if err := json.NewDecoder(resp.Body).Decode(&frames); err != nil { + t.Fatalf("decode: %v", err) + } + if len(frames) != 1 { + t.Fatalf("frames = %d, want 1", len(frames)) + } + f := frames[0] + if f["type"] != evMessageEnd || f["sessionId"] != "s1" || f["seq"].(float64) != 1 || f["v"].(float64) != 1 { + t.Fatalf("frame = %v", f) + } + msg := f["message"].(map[string]any) + if msg["text"] != "hi" { + t.Fatalf("message = %v", msg) + } +} + +func TestAPISpawnValidation(t *testing.T) { + ts, _ := newTestServer(t) + req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/spawn", + strings.NewReader(`{"repo":"no-slash"}`)) + req.Header.Set("Authorization", "Bearer "+testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("spawn: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("bad repo → %d, want 400", resp.StatusCode) + } +} + +func TestAPIRecoverMiddleware(t *testing.T) { + // A handler that panics must yield 500, not kill the server. + handler := recoverMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + panic("boom") + })) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil)) + if rec.Code != http.StatusInternalServerError { + t.Fatalf("panic → %d, want 500", rec.Code) + } + if !strings.Contains(rec.Body.String(), "error") { + t.Fatalf("body = %q", rec.Body.String()) + } +} diff --git a/daemon/docker.go b/daemon/docker.go new file mode 100644 index 0000000..0012d9e --- /dev/null +++ b/daemon/docker.go @@ -0,0 +1,539 @@ +package main + +// docker.go — worker container spawner (clone → ensure image → run). + +import ( + "archive/tar" + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "sync" + "time" + + "github.com/docker/docker/api/types/build" + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" + "github.com/docker/docker/api/types/image" + "github.com/docker/docker/api/types/volume" + "github.com/docker/docker/client" +) + +// Spawn job states and docker wiring constants. +const ( + stateCloning string = "cloning" + stateBuilding string = "building" + stateCreating string = "creating" + stateRunning string = "running" + stateError string = "error" + + imageRefWorker string = "lvmh-worker:latest" + labelSession string = "lvmh.session" + volumeRepoPrefix string = "lvmh-repo-" + volumeSessions string = "lvmh-sessions" + workspaceMount string = "/workspace" + sessionsMount string = "/pi-sessions" + modelsMountTarget string = "/root/.pi/agent/models.json" + + envWorkerDockerfile string = "LVMH_WORKER_DOCKERFILE" + defaultDockerfile string = "/app/build/docker/worker.Dockerfile" + envWorkerContext string = "LVMH_WORKER_CONTEXT" + envRepoDir string = "LVMH_REPO_DIR" + defaultRepoDir string = "/data/repos" + envContainerLVMHURL string = "LVMH_CONTAINER_LVMH_URL" + defaultContainerURL string = "ws://lvmh:8686/agent/ws" + envContainerNetwork string = "LVMH_CONTAINER_NETWORK" + defaultNetwork string = "lvmh-net" + envWorkerModels string = "LVMH_WORKER_MODELS" + defaultWorkerModels string = "/app/build/docker/worker-models.json" + envProviderAPIKey string = "ZAI_RENAUD_API_KEY" + envLVMHSessionID string = "LVMH_SESSION_ID" + stopTimeoutSeconds int = 10 + buildContextReadLimit int64 = 64 << 20 +) + +var errNoContainer = errors.New("no container for session") + +var repoPathRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+(/[A-Za-z0-9_.-]+)+$`) + +// validRepoPath accepts "group/project" style paths (at least two segments). +func validRepoPath(repo string) bool { return repoPathRe.MatchString(repo) } + +func repoSlug(repo string) string { return strings.ReplaceAll(repo, "/", "-") } + +// newUUID returns a random RFC 4122 v4 UUID string. +func newUUID() string { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + // crypto/rand failure is fatal-grade; never expected in practice. + panic("newUUID: crypto/rand unavailable: " + err.Error()) + } + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + dst := make([]byte, 36) + hex.Encode(dst[0:8], b[0:4]) + dst[8] = '-' + hex.Encode(dst[9:13], b[4:6]) + dst[13] = '-' + hex.Encode(dst[14:18], b[6:8]) + dst[18] = '-' + hex.Encode(dst[19:23], b[8:10]) + dst[23] = '-' + hex.Encode(dst[24:36], b[10:16]) + return string(dst) +} + +// SpawnJob is one async spawn pipeline run, exposed via /api/spawn/status. +type SpawnJob struct { + Repo string `json:"repo"` + State string `json:"state"` + ContainerID string `json:"containerId"` + SessionID string `json:"sessionId,omitempty"` + Message string `json:"message,omitempty"` + UpdatedAt int64 `json:"-"` +} + +// SpawnResult is the immediate POST /api/spawn response. +type SpawnResult struct { + SessionID string `json:"sessionId"` + ContainerID string `json:"containerId"` +} + +// Spawner owns the docker client, per-repo clone serialization and job state. +type Spawner struct { + store *Store + hub *Hub + cli *client.Client + baseURL string // gitlab base, for clone URLs + + reposDir string + dockerfile string + buildContext string + containerURL string + network string + modelsPath string + + mu sync.Mutex + jobs map[string]*SpawnJob // keyed by sessionId + slugLocks map[string]*sync.Mutex +} + +func NewSpawner(store *Store, hub *Hub, baseURL string) (*Spawner, error) { + cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) + if err != nil { + return nil, fmt.Errorf("docker client: %w", err) + } + dockerfile := envOr(envWorkerDockerfile, defaultDockerfile) + return &Spawner{ + store: store, + hub: hub, + cli: cli, + baseURL: strings.TrimRight(baseURL, "/"), + reposDir: envOr(envRepoDir, defaultRepoDir), + dockerfile: dockerfile, + buildContext: envOr(envWorkerContext, filepath.Dir(filepath.Dir(dockerfile))), + containerURL: envOr(envContainerLVMHURL, defaultContainerURL), + network: envOr(envContainerNetwork, defaultNetwork), + modelsPath: envOr(envWorkerModels, defaultWorkerModels), + jobs: make(map[string]*SpawnJob), + slugLocks: make(map[string]*sync.Mutex), + }, nil +} + +func envOr(key, def string) string { + if v := strings.TrimSpace(os.Getenv(key)); v != "" { + return v + } + return def +} + +// JobsSnapshot returns current spawn jobs for /api/spawn/status and /ws pushes. +func (s *Spawner) JobsSnapshot() []SpawnJob { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]SpawnJob, 0, len(s.jobs)) + for _, j := range s.jobs { + out = append(out, *j) + } + return out +} + +func (s *Spawner) setJob(sessionID, repo, state, containerID, message string) { + s.mu.Lock() + j, ok := s.jobs[sessionID] + if !ok { + j = &SpawnJob{SessionID: sessionID, Repo: repo} + s.jobs[sessionID] = j + } + j.Repo = repo + j.State = state + j.ContainerID = containerID + j.Message = message + j.UpdatedAt = time.Now().UnixMilli() + s.mu.Unlock() + s.hub.BroadcastSpawnStatus() +} + +// Start launches the async spawn pipeline and returns the new sessionId. +func (s *Spawner) Start(ctx context.Context, repo, branch string) (SpawnResult, error) { + if _, err := s.imageExists(ctx); err != nil { + return SpawnResult{}, fmt.Errorf("docker unavailable: %w", err) + } + exists, err := s.imageExists(ctx) + if err == nil && !exists { + if _, statErr := os.Stat(s.dockerfile); statErr != nil { + return SpawnResult{}, fmt.Errorf( + "image %s not found and no worker Dockerfile at %s (set %s or run `make worker-image`)", + imageRefWorker, s.dockerfile, envWorkerDockerfile) + } + } + sessionID := newUUID() + s.setJob(sessionID, repo, stateCloning, "", "") + go s.runJob(repo, branch, sessionID) + return SpawnResult{SessionID: sessionID}, nil +} + +func (s *Spawner) imageExists(ctx context.Context) (bool, error) { + args := filters.NewArgs(filters.Arg("reference", imageRefWorker)) + summaries, err := s.cli.ImageList(ctx, image.ListOptions{Filters: args}) + if err != nil { + return false, err + } + return len(summaries) > 0, nil +} + +func (s *Spawner) slugLock(slug string) *sync.Mutex { + s.mu.Lock() + defer s.mu.Unlock() + l, ok := s.slugLocks[slug] + if !ok { + l = &sync.Mutex{} + s.slugLocks[slug] = l + } + return l +} + +// runJob is the async clone→build→create→start pipeline. +func (s *Spawner) runJob(repo, branch, sessionID string) { + slug := repoSlug(repo) + lock := s.slugLock(slug) + lock.Lock() + defer lock.Unlock() + + if err := s.cloneOrUpdate(repo, branch, slug); err != nil { + s.setJob(sessionID, repo, stateError, "", err.Error()) + return + } + s.setJob(sessionID, repo, stateBuilding, "", "") + if err := s.ensureImage(context.Background()); err != nil { + s.setJob(sessionID, repo, stateError, "", err.Error()) + return + } + s.setJob(sessionID, repo, stateCreating, "", "") + containerID, err := s.createAndStart(context.Background(), repo, slug, sessionID) + if err != nil { + s.setJob(sessionID, repo, stateError, "", err.Error()) + return + } + if err := s.store.UpsertContainer(sessionID, containerID, repo); err != nil { + s.setJob(sessionID, repo, stateError, containerID, "container started but not persisted: "+err.Error()) + return + } + s.setJob(sessionID, repo, stateRunning, containerID, "") +} + +// cloneOrUpdate clones the repo into reposDir/ or fast-forwards it. +func (s *Spawner) cloneOrUpdate(repo, branch, slug string) error { + dir := filepath.Join(s.reposDir, slug) + cloneURL, err := s.cloneURL(repo) + if err != nil { + return err + } + if st, err := os.Stat(filepath.Join(dir, ".git")); err == nil && st.IsDir() { + if err := gitRun(dir, "pull", "--ff-only"); err != nil { + return fmt.Errorf("git pull %s: %w", repo, err) + } + return nil + } + if err := os.MkdirAll(s.reposDir, 0o755); err != nil { + return err + } + args := []string{"clone"} + if branch != "" { + args = append(args, "--branch", branch) + } + args = append(args, "--", cloneURL, dir) + if err := gitRun("", args...); err != nil { + return fmt.Errorf("git clone %s: %w", repo, err) + } + return nil +} + +// cloneURL builds an authenticated https clone URL when a PAT is stored. +func (s *Spawner) cloneURL(repo string) (string, error) { + u, err := url.Parse(s.baseURL + "/" + repo + ".git") + if err != nil { + return "", err + } + if token, ok, _ := s.store.GetSetting(settingGitLabToken); ok && token != "" && u.User == nil { + u.User = url.UserPassword("oauth2", token) + } + return u.String(), nil +} + +func gitRun(dir string, args ...string) error { + cmd := exec.Command("git", args...) + if dir != "" { + cmd.Dir = dir + } + var out bytes.Buffer + cmd.Stdout = &out + cmd.Stderr = &out + if err := cmd.Run(); err != nil { + msg := strings.TrimSpace(out.String()) + if len(msg) > 500 { + msg = msg[len(msg)-500:] + } + if msg != "" { + return fmt.Errorf("%w: %s", err, msg) + } + return err + } + return nil +} + +// ensureImage builds lvmh-worker:latest when missing. +func (s *Spawner) ensureImage(ctx context.Context) error { + exists, err := s.imageExists(ctx) + if err != nil { + return err + } + if exists { + return nil + } + if _, err := os.Stat(s.dockerfile); err != nil { + return fmt.Errorf("worker Dockerfile missing at %s: %w", s.dockerfile, err) + } + relDockerfile, err := filepath.Rel(s.buildContext, s.dockerfile) + if err != nil { + return err + } + var buf bytes.Buffer + if err := tarDir(&buf, s.buildContext); err != nil { + return fmt.Errorf("build context %s: %w", s.buildContext, err) + } + resp, err := s.cli.ImageBuild(ctx, &buf, build.ImageBuildOptions{ + Tags: []string{imageRefWorker}, + Dockerfile: relDockerfile, + Remove: true, + }) + if err != nil { + return fmt.Errorf("docker build: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, buildContextReadLimit)) + if msg := extractBuildError(body); msg != "" { + return fmt.Errorf("docker build failed: %s", msg) + } + return nil +} + +// extractBuildError scans the build output stream for an "error" field. +func extractBuildError(body []byte) string { + for _, line := range bytes.Split(body, []byte("\n")) { + var msg struct { + Error string `json:"error"` + ErrorDetail struct { + Message string `json:"message"` + } `json:"errorDetail"` + } + if json.Unmarshal(line, &msg) != nil { + continue + } + if msg.Error != "" || msg.ErrorDetail.Message != "" { + if msg.ErrorDetail.Message != "" { + return msg.ErrorDetail.Message + } + return msg.Error + } + } + return "" +} + +// createAndStart provisions volumes, creates and starts the worker container. +func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID string) (string, error) { + repoVolume := volumeRepoPrefix + slug + fresh, err := s.ensureRepoVolume(ctx, repoVolume) + if err != nil { + return "", err + } + if fresh { + if err := s.seedVolume(ctx, slug, repoVolume); err != nil { + return "", fmt.Errorf("seed %s: %w", repoVolume, err) + } + } + if _, err := s.cli.VolumeCreate(ctx, volume.CreateOptions{Name: volumeSessions}); err != nil { + return "", fmt.Errorf("volume %s: %w", volumeSessions, err) + } + + binds := []string{repoVolume + ":" + workspaceMount, volumeSessions + ":" + sessionsMount} + if _, err := os.Stat(s.modelsPath); err == nil { + binds = append(binds, s.modelsPath+":"+modelsMountTarget+":ro") + } + cfg := &container.Config{ + Image: imageRefWorker, + Env: []string{ + envProviderAPIKey + "=" + os.Getenv(envProviderAPIKey), + envToken + "=" + daemonToken, + "LVMH_URL=" + s.containerURL, + envLVMHSessionID + "=" + sessionID, + }, + Labels: map[string]string{labelSession: sessionID}, + } + hostCfg := &container.HostConfig{ + Binds: binds, + NetworkMode: container.NetworkMode(s.network), + AutoRemove: false, + } + name := "lvmh-agent-" + strings.ReplaceAll(sessionID, "-", "")[:12] + created, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, name) + if err != nil { + return "", fmt.Errorf("docker create: %w", err) + } + if err := s.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil { + s.removeContainer(context.Background(), created.ID) + return "", fmt.Errorf("docker start: %w", err) + } + return created.ID, nil +} + +// ensureRepoVolume creates the per-repo volume; reports whether it is fresh. +func (s *Spawner) ensureRepoVolume(ctx context.Context, name string) (bool, error) { + if _, err := s.cli.VolumeInspect(ctx, name); err == nil { + return false, nil + } + if _, err := s.cli.VolumeCreate(ctx, volume.CreateOptions{Name: name}); err != nil { + return false, fmt.Errorf("volume %s: %w", name, err) + } + return true, nil +} + +// seedVolume copies the cloned repo into the fresh named volume via a +// one-shot container from the worker image (no extra images needed). +func (s *Spawner) seedVolume(ctx context.Context, slug, repoVolume string) error { + repoDir := filepath.Join(s.reposDir, slug) + cfg := &container.Config{ + Image: imageRefWorker, + Cmd: []string{"sh", "-c", "cp -a /src/. " + workspaceMount + "/"}, + } + hostCfg := &container.HostConfig{ + Binds: []string{ + repoVolume + ":" + workspaceMount, + repoDir + ":/src:ro", + }, + } + created, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, "lvmh-seed-"+slug) + if err != nil { + return err + } + defer s.removeContainer(context.Background(), created.ID) + if err := s.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil { + return err + } + waitCh, errCh := s.cli.ContainerWait(ctx, created.ID, container.WaitConditionNotRunning) + select { + case <-waitCh: + return nil + case err := <-errCh: + return err + case <-ctx.Done(): + return ctx.Err() + } +} + +func (s *Spawner) removeContainer(ctx context.Context, id string) { + _ = s.cli.ContainerRemove(ctx, id, container.RemoveOptions{Force: true}) +} + +// RemoveSession stops and removes the container spawned for a session. +func (s *Spawner) RemoveSession(ctx context.Context, sessionID string) error { + row, ok, err := s.store.GetContainer(sessionID) + if err != nil { + return err + } + if !ok { + return errNoContainer + } + stopCtx, cancel := context.WithTimeout(context.Background(), time.Duration(stopTimeoutSeconds)*time.Second) + if err := s.cli.ContainerStop(stopCtx, row.ContainerID, container.StopOptions{}); err != nil { + cancel() + return fmt.Errorf("docker stop: %w", err) + } + cancel() + s.removeContainer(ctx, row.ContainerID) + if err := s.store.DeleteContainer(sessionID); err != nil { + return err + } + s.mu.Lock() + if j, ok := s.jobs[sessionID]; ok { + j.State = stateError + j.Message = "container removed" + j.UpdatedAt = time.Now().UnixMilli() + } + s.mu.Unlock() + s.hub.BroadcastSpawnStatus() + return nil +} + +// tarDir writes a tar archive of dir (excluding .git-heavy junk) into w. +func tarDir(w io.Writer, dir string) error { + tw := tar.NewWriter(w) + defer tw.Close() + return filepath.Walk(dir, func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + rel, err := filepath.Rel(dir, path) + if err != nil { + return err + } + if rel == "." { + return nil + } + name := filepath.ToSlash(rel) + hdr, err := tar.FileInfoHeader(info, "") + if err != nil { + return err + } + hdr.Name = name + if info.IsDir() { + hdr.Name += "/" + } + hdr.Uname = "root" + hdr.Gname = "root" + if err := tw.WriteHeader(hdr); err != nil { + return err + } + if info.Mode().IsRegular() { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + if _, err := io.Copy(tw, f); err != nil { + return err + } + } + return nil + }) +} diff --git a/daemon/gitlab.go b/daemon/gitlab.go new file mode 100644 index 0000000..f1dbc68 --- /dev/null +++ b/daemon/gitlab.go @@ -0,0 +1,184 @@ +package main + +// gitlab.go — GitLab PAT management and project listing. + +import ( + "context" + "crypto/tls" + "encoding/json" + "errors" + "fmt" + "net/http" + "sort" + "strings" + "time" +) + +// Settings keys and GitLab defaults. +const ( + settingGitLabToken string = "gitlab_pat" + projectsPerPage int = 50 + gitlabTimeout time.Duration = 15 * time.Second +) + +// GitLabError marks upstream GitLab failures (mapped to 502 by the API). +type GitLabError struct{ msg string } + +func (e *GitLabError) Error() string { return e.msg } + +var errNotConnected = errors.New("gitlab not connected") + +// GitLabRepo is the /api/gitlab/repos item shape (protocol field names). +type GitLabRepo struct { + Path string `json:"path"` + Name string `json:"name"` + Namespace string `json:"namespace"` + LastActivityAt string `json:"lastActivityAt"` + WebURL string `json:"webUrl"` + DefaultBranch string `json:"defaultBranch"` +} + +// gitlabProject is the subset of the upstream projects API we map from. +type gitlabProject struct { + PathWithNamespace string `json:"path_with_namespace"` + Name string `json:"name"` + Namespace struct { + Path string `json:"path"` + } `json:"namespace"` + LastActivityAt string `json:"last_activity_at"` + WebURL string `json:"web_url"` + DefaultBranch string `json:"default_branch"` +} + +// GitLab stores/validates the PAT in the settings table and lists projects. +type GitLab struct { + store *Store + baseURL string + insecure bool +} + +func NewGitLab(store *Store, baseURL string) *GitLab { + return &GitLab{store: store, baseURL: strings.TrimRight(baseURL, "/")} +} + +// NewGitLabWithClient is the test seam: baseURL + transport overrides. +func NewGitLabWithClient(store *Store, baseURL string, insecureSkipVerify bool) *GitLab { + return &GitLab{store: store, baseURL: strings.TrimRight(baseURL, "/"), insecure: insecureSkipVerify} +} + +func (g *GitLab) httpClient() *http.Client { + client := &http.Client{Timeout: gitlabTimeout} + if g.insecure { + client.Transport = &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // test seam only + } + } + return client +} + +// Status reports connection state; never includes the PAT. +func (g *GitLab) Status() map[string]any { + out := map[string]any{"connected": false, "baseUrl": g.baseURL} + token, ok, err := g.store.GetSetting(settingGitLabToken) + if err != nil || !ok || token == "" { + return out + } + out["connected"] = true + if username, ok, _ := g.store.GetSetting(settingGitLabUsername); ok { + out["username"] = username + } + return out +} + +const settingGitLabUsername string = "gitlab_username" + +// Connect validates the PAT against /api/v4/user and stores it. +func (g *GitLab) Connect(ctx context.Context, token string) (string, error) { + var user struct { + Username string `json:"username"` + } + if err := g.do(ctx, "/api/v4/user", token, &user); err != nil { + return "", err + } + if user.Username == "" { + return "", &GitLabError{msg: "gitlab returned no username for token"} + } + if err := g.store.SetSetting(settingGitLabToken, token); err != nil { + return "", err + } + if err := g.store.SetSetting(settingGitLabUsername, user.Username); err != nil { + return "", err + } + return user.Username, nil +} + +// Disconnect drops the stored PAT. +func (g *GitLab) Disconnect() error { + if err := g.store.DeleteSetting(settingGitLabToken); err != nil { + return err + } + return g.store.DeleteSetting(settingGitLabUsername) +} + +// token returns the stored PAT or errNotConnected. +func (g *GitLab) token() (string, error) { + token, ok, err := g.store.GetSetting(settingGitLabToken) + if err != nil { + return "", err + } + if !ok || token == "" { + return "", errNotConnected + } + return token, nil +} + +// Repos lists member projects sorted by most recent activity. +func (g *GitLab) Repos(ctx context.Context) ([]GitLabRepo, error) { + token, err := g.token() + if err != nil { + return nil, err + } + var projects []gitlabProject + path := fmt.Sprintf("/api/v4/projects?membership=true&order_by=last_activity_at&per_page=%d", projectsPerPage) + if err := g.do(ctx, path, token, &projects); err != nil { + return nil, err + } + repos := make([]GitLabRepo, 0, len(projects)) + for _, p := range projects { + if p.PathWithNamespace == "" { + continue + } + repos = append(repos, GitLabRepo{ + Path: p.PathWithNamespace, + Name: p.Name, + Namespace: p.Namespace.Path, + LastActivityAt: p.LastActivityAt, + WebURL: p.WebURL, + DefaultBranch: p.DefaultBranch, + }) + } + sort.SliceStable(repos, func(i, j int) bool { + return repos[i].LastActivityAt > repos[j].LastActivityAt + }) + return repos, nil +} + +func (g *GitLab) do(ctx context.Context, path, token string, out any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.baseURL+path, nil) + if err != nil { + return err + } + req.Header.Set("Private-Token", token) + resp, err := g.httpClient().Do(req) + if err != nil { + return &GitLabError{msg: fmt.Sprintf("gitlab request failed: %v", err)} + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return &GitLabError{msg: fmt.Sprintf("gitlab %s returned %d", path, resp.StatusCode)} + } + if err := json.NewDecoder(resp.Body).Decode(out); err != nil { + return &GitLabError{msg: fmt.Sprintf("gitlab %s: decode: %v", path, err)} + } + return nil +} diff --git a/daemon/gitlab_test.go b/daemon/gitlab_test.go new file mode 100644 index 0000000..51b3f75 --- /dev/null +++ b/daemon/gitlab_test.go @@ -0,0 +1,123 @@ +package main + +// gitlab_test.go — PAT connect + project mapping against httptest upstream. + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func newGitLabUpstream(t *testing.T) (*httptest.Server, *GitLab) { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("/api/v4/user", func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Private-Token") != "pat-good" { + w.WriteHeader(http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"username":"alice"}`)) + }) + mux.HandleFunc("/api/v4/projects", func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Private-Token") != "pat-good" { + w.WriteHeader(http.StatusUnauthorized) + return + } + q := r.URL.Query() + if q.Get("membership") != "true" || q.Get("order_by") != "last_activity_at" { + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`[ + {"path_with_namespace":"team/b","name":"B","namespace":{"path":"team"}, + "last_activity_at":"2024-02-02T00:00:00Z","web_url":"https://gl/team/b","default_branch":"main"}, + {"path_with_namespace":"team/a","name":"A","namespace":{"path":"team"}, + "last_activity_at":"2024-03-03T00:00:00Z","web_url":"https://gl/team/a","default_branch":"trunk"} + ]`)) + }) + ts := httptest.NewServer(mux) + t.Cleanup(ts.Close) + store := openTestStore(t) + return ts, NewGitLab(store, ts.URL) +} + +func TestGitLabConnectValidatesAndStores(t *testing.T) { + _, gl := newGitLabUpstream(t) + + if status := gl.Status(); status["connected"] != false { + t.Fatalf("status before connect = %v", status) + } + + if _, err := gl.Connect(context.Background(), "pat-bad"); err == nil { + t.Fatal("bad PAT should fail validation") + } + + username, err := gl.Connect(context.Background(), "pat-good") + if err != nil { + t.Fatalf("connect: %v", err) + } + if username != "alice" { + t.Fatalf("username = %q", username) + } + + status := gl.Status() + if status["connected"] != true || status["username"] != "alice" { + t.Fatalf("status after connect = %v", status) + } + if strings.Contains(mustJSON(t, status), "pat-good") { + t.Fatal("status must never expose the PAT") + } + + if err := gl.Disconnect(); err != nil { + t.Fatalf("disconnect: %v", err) + } + if status := gl.Status(); status["connected"] != false { + t.Fatalf("status after disconnect = %v", status) + } +} + +func mustJSON(t *testing.T, v any) string { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return string(b) +} + +func TestGitLabReposMappingAndSorting(t *testing.T) { + _, gl := newGitLabUpstream(t) + + if _, err := gl.Repos(context.Background()); err == nil || !strings.Contains(err.Error(), "not connected") { + t.Fatalf("repos without connect = %v, want not-connected error", err) + } + + if _, err := gl.Connect(context.Background(), "pat-good"); err != nil { + t.Fatalf("connect: %v", err) + } + repos, err := gl.Repos(context.Background()) + if err != nil { + t.Fatalf("repos: %v", err) + } + if len(repos) != 2 { + t.Fatalf("repos = %d, want 2", len(repos)) + } + want := GitLabRepo{ + Path: "team/a", Name: "A", Namespace: "team", + LastActivityAt: "2024-03-03T00:00:00Z", WebURL: "https://gl/team/a", DefaultBranch: "trunk", + } + if repos[0] != want { + t.Fatalf("first repo = %+v, want %+v (sorted by activity desc)", repos[0], want) + } + encoded := mustJSON(t, repos[0]) + for _, key := range []string{`"path":"team/a"`, `"namespace":"team"`, `"webUrl":"https://gl/team/a"`, `"defaultBranch":"trunk"`} { + if !strings.Contains(encoded, key) { + t.Fatalf("repo JSON %s missing %s", encoded, key) + } + } +} diff --git a/daemon/go.mod b/daemon/go.mod new file mode 100644 index 0000000..714304a --- /dev/null +++ b/daemon/go.mod @@ -0,0 +1,47 @@ +module lvmh-daemon + +go 1.25.0 + +require ( + github.com/docker/docker v28.5.2+incompatible + github.com/gorilla/websocket v1.5.3 + modernc.org/sqlite v1.56.0 +) + +require ( + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/errdefs v1.0.0 // indirect + github.com/containerd/errdefs/pkg v0.3.0 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/go-connections v0.8.1 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/felixge/httpsnoop v1.1.0 // indirect + github.com/go-logr/logr v1.4.4 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.24 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/sys/atomicwriter v0.1.0 // indirect + github.com/moby/term v0.5.2 // indirect + github.com/morikuni/aec v1.1.0 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.70.0 // indirect + go.opentelemetry.io/otel v1.45.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0 // indirect + go.opentelemetry.io/otel/metric v1.45.0 // indirect + go.opentelemetry.io/otel/trace v1.45.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/time v0.15.0 // indirect + gotest.tools/v3 v3.5.2 // indirect + modernc.org/libc v1.74.4 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.11.0 // indirect +) diff --git a/daemon/go.sum b/daemon/go.sum new file mode 100644 index 0000000..4b5d1be --- /dev/null +++ b/daemon/go.sum @@ -0,0 +1,147 @@ +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= +github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM= +github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/go-connections v0.8.1 h1:JibmG5hULs5qXSr/cp/w3Pw5fZuStt4MOHMUExb29/M= +github.com/docker/go-connections v0.8.1/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= +github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= +github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFebv6EsYotImrt/Ppc5cXIriCSo= +github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3/go.mod h1:jl5iWTm0/hd5PjEYEOuwAJ57L/CibdZfrqZ5XA5GrCk= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= +github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w3A14Sw= +github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs= +github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= +github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/morikuni/aec v1.1.0 h1:vBBl0pUnvi/Je71dsRrhMBtreIqNMYErSAbEeb8jrXQ= +github.com/morikuni/aec v1.1.0/go.mod h1:xDRgiq/iw5l+zkao76YTKzKttOp2cwPEne25HDkJnBw= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= +github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.70.0 h1:LMuyCAyfalSjDyjdC65nK6N0zoTT63+E/u95X0JovZI= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.70.0/go.mod h1:085m8qbm4hgc8rZWGDEa4vmyyo2c3nPxUslYUKUIU04= +go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= +go.opentelemetry.io/otel v1.45.0/go.mod h1:XZxIqPapzEYnhNSScF5DIqXhm/rYi0FzCe2XddAwZfQ= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 h1:QRefszxJmfPdjXUUm3j6iDzY03mTPXMjqErFqQ67vUg= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0/go.mod h1:Tiz03lTBVBrm7eWZBOidzEaYaJa8tjwGUGv6d8mlTyk= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0 h1:QBajQ2SrwQijzHyZbQlPsuIzpl/ll8DY6wPWsajeGcI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.45.0/go.mod h1:08ZQLjrPLQ6R4kAXvuOvODEer5Yh4CoFvll5qB2BCI8= +go.opentelemetry.io/otel/metric v1.45.0 h1:7Eg1uH7CJ5cXv9is6tnBe1FI6rj1nwUdbFypRm3br/M= +go.opentelemetry.io/otel/metric v1.45.0/go.mod h1:HAPbm1nd3p1PmFH7v2dR+6BjXxw+Lq4a2+pndMAm08s= +go.opentelemetry.io/otel/sdk v1.45.0 h1:4VVSMgQ83dUgW2aoX5f6JgLvHwIvzcuLnF9lUdCSpCw= +go.opentelemetry.io/otel/sdk v1.45.0/go.mod h1:Sr40LgXV7DsKMMJMKOhUWOgMWTfAaqvm2kF0g7ilwuA= +go.opentelemetry.io/otel/sdk/metric v1.45.0 h1:oVFszMfyj1Am6s24Vtc7wBb8BKLcwepJjNEYILuiE3o= +go.opentelemetry.io/otel/sdk/metric v1.45.0/go.mod h1:vUWUxDZvu1WVRj8JA8S0AdhsPrZoDpA2DdZauIh4mDA= +go.opentelemetry.io/otel/trace v1.45.0 h1:l/mP6Uv7oNO7/TblbhpbgMidxhq1uO/rPsikOyVhxag= +go.opentelemetry.io/otel/trace v1.45.0/go.mod h1:qoJJA2xNMnxRrdISU/kLtfUH2wNeQbiv+jhs/CxI8bc= +go.opentelemetry.io/proto/otlp v1.11.0 h1:5rrYs0Ykyj50sdU/JU0x8etU+LubXWb+gED6TbEdMIk= +go.opentelemetry.io/proto/otlp v1.11.0/go.mod h1:SmVizdCOAm3XBtG1g1NnOdhW6jtddT72hLMhv8VwA8E= +golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= +golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= +golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= +golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= +golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= +google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d h1:FarXi840EJWSHYTN3ERkADbPWjl307+FGrA22KAVjjc= +google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d/go.mod h1:K/+WGbmBY7aNW1HDw1fJnKYo10i0DkAX6pows00dLig= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d h1:IL4hdHzcUv2l/gcg98/Rj3FbtE6axwqslOW8SW0C+S0= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ= +google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= +gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +modernc.org/cc/v4 v4.29.1 h1:MKgdCV3WykTSPqpVrnxdEDS0HEd2FHpKZDzxzU5LyeI= +modernc.org/cc/v4 v4.29.1/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= +modernc.org/ccgo/v4 v4.34.6 h1:sBgfIwyN0TQ9C5hwIeuqyeAKyMWnbvj2fvpF4L11uzU= +modernc.org/ccgo/v4 v4.34.6/go.mod h1:SZ8YcN9NG7XVsQYdm6jYBvi8PQP1qi+kqB6OhjqI3Fk= +modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= +modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/gc/v3 v3.1.4 h1:2g65LGVSmFQrXeITAw97x7hCRvZFcyE1uDP+7Vng7JI= +modernc.org/gc/v3 v3.1.4/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.74.4 h1:fX1Omw4o2/1C2iRkkIsrQTasJQldLhRmuPreXLoWs9k= +modernc.org/libc v1.74.4/go.mod h1:eeQAS9W3sZeKYMFubydxJpII9ybHWshk+7or7bLG9co= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg= +modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= +modernc.org/sqlite v1.56.0 h1:/D8e2RfFqoy/Zc6PuC76U28zFwmI/sYx1Kjm4yEn9e0= +modernc.org/sqlite v1.56.0/go.mod h1:yCJ2cmAaIkHQ25oXWrF8H4O1lIfPYPR26yCEDj2P3pQ= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/daemon/hub.go b/daemon/hub.go new file mode 100644 index 0000000..c926317 --- /dev/null +++ b/daemon/hub.go @@ -0,0 +1,621 @@ +package main + +// hub.go — agent websocket hub (/agent/ws) and web websocket hub (/ws). + +import ( + "encoding/json" + "errors" + "log" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +// Event types on the agent wire (and control frames on the web wire). +const ( + evHello string = "hello" + evWelcome string = "welcome" + evMessageStart string = "message_start" + evMessageUpdate string = "message_update" + evMessageEnd string = "message_end" + evToolExecStart string = "tool_execution_start" + evToolExecUpdate string = "tool_execution_update" + evToolExecEnd string = "tool_execution_end" + evAgentStart string = "agent_start" + evAgentEnd string = "agent_end" + evAgentSettled string = "agent_settled" + evSessionInfo string = "session_info" + evBye string = "bye" + evPrompt string = "prompt" + evAbort string = "abort" + frameSessionList string = "session_list" + frameEvents string = "events" + frameSpawnStatus string = "spawn_status" + frameSubscribe string = "subscribe" + frameUnsubscribe string = "unsubscribe" + evPersistedSession string = "session" // payload key of hello/session_info +) + +// Wire tuning constants. +const ( + maxFrameSize int64 = 1 << 20 // 1 MiB per inbound frame + agentSendQueue int = 32 + webSendQueue int = 128 + webFlushInterval time.Duration = 40 * time.Millisecond + webMaxPending int = 4096 // drop slow clients beyond this backlog + promptSendTimeout time.Duration = 3 * time.Second + writeWait time.Duration = 5 * time.Second + daemonVersion int = 1 // envelope "v" +) + +// ErrOffline is returned when a prompt/abort targets a session with no live agent WS. +var ErrOffline = errors.New("session offline") + +// frame is one decoded agent wire frame (envelope fields flattened with payload). +type frame struct { + raw []byte + typ string + sessionID string + seq int64 + ts int64 +} + +func decodeFrame(data []byte) (frame, error) { + var fields map[string]json.RawMessage + if err := json.Unmarshal(data, &fields); err != nil { + return frame{}, err + } + f := frame{raw: data} + if raw, ok := fields["type"]; ok { + _ = json.Unmarshal(raw, &f.typ) + } + if raw, ok := fields["sessionId"]; ok { + _ = json.Unmarshal(raw, &f.sessionID) + } + if raw, ok := fields["seq"]; ok { + _ = json.Unmarshal(raw, &f.seq) + } + if raw, ok := fields["ts"]; ok { + _ = json.Unmarshal(raw, &f.ts) + } + return f, nil +} + +// SessionView is the /api/sessions item shape. +type SessionView struct { + ID string `json:"id"` + Name *string `json:"name"` + Cwd string `json:"cwd"` + Model string `json:"model"` + Provider string `json:"provider"` + Agent bool `json:"agent"` + Repo *string `json:"repo"` + StartedAt int64 `json:"startedAt"` + Online bool `json:"online"` + LastEventAt int64 `json:"lastEventAt"` +} + +// pendingEvent is one agent frame queued for a web subscriber. +type pendingEvent struct { + sessionID string + seq int64 + raw json.RawMessage +} + +// webClient is one browser websocket with batched outbound delivery. +type webClient struct { + hub *Hub + conn *websocket.Conn + + mu sync.Mutex + sub string // subscribed sessionId, "" when none + control [][]byte + events []pendingEvent + dropped bool + + closeOnce sync.Once + done chan struct{} +} + +func (c *webClient) drop() { + c.closeOnce.Do(func() { + close(c.done) + _ = c.conn.Close() + }) +} + +func (c *webClient) deliverControl(b []byte) { + c.mu.Lock() + defer c.mu.Unlock() + if c.dropped { + return + } + if len(c.control)+len(c.events) >= webMaxPending { + c.dropped = true + go c.drop() + return + } + c.control = append(c.control, b) +} + +func (c *webClient) deliverEvent(e pendingEvent) { + c.mu.Lock() + defer c.mu.Unlock() + if c.dropped { + return + } + if len(c.control)+len(c.events) >= webMaxPending { + c.dropped = true + go c.drop() + return + } + c.events = append(c.events, e) +} + +// writePump flushes queued frames every webFlushInterval, batching events of +// the subscribed session into single frames. Never blocks the hub. +func (c *webClient) writePump() { + ticker := time.NewTicker(webFlushInterval) + defer ticker.Stop() + for { + select { + case <-c.done: + return + case <-ticker.C: + } + c.mu.Lock() + control := c.control + c.control = nil + events := c.events + c.events = nil + c.mu.Unlock() + for _, b := range control { + _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := c.conn.WriteMessage(websocket.TextMessage, b); err != nil { + c.hub.dropWeb(c) + return + } + } + if len(events) == 0 { + continue + } + items := make([]json.RawMessage, 0, len(events)) + after := events[0].seq - 1 + for _, e := range events { + items = append(items, e.raw) + } + payload, err := json.Marshal(map[string]any{ + "type": frameEvents, + "sessionId": events[0].sessionID, + "after": after, + "events": items, + }) + if err != nil { + continue + } + _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil { + c.hub.dropWeb(c) + return + } + } +} + +// readPump consumes subscribe/unsubscribe frames; malformed input never kills the server. +func (c *webClient) readPump() { + defer c.hub.dropWeb(c) + c.conn.SetReadLimit(maxFrameSize) + for { + _, data, err := c.conn.ReadMessage() + if err != nil { + return + } + var msg map[string]any + if err := json.Unmarshal(data, &msg); err != nil { + continue + } + typ, _ := msg["type"].(string) + sessionID, _ := msg["sessionId"].(string) + c.mu.Lock() + switch typ { + case frameSubscribe: + c.sub = sessionID + case frameUnsubscribe: + c.sub = "" + } + c.mu.Unlock() + } +} + +func (c *webClient) subscription() string { + c.mu.Lock() + defer c.mu.Unlock() + return c.sub +} + +// agentConn is one plugin websocket bound to a session after hello. +type agentConn struct { + hub *Hub + conn *websocket.Conn + sessionID string + send chan []byte + done chan struct{} + closeOnce sync.Once +} + +func (a *agentConn) drop() { + a.closeOnce.Do(func() { + close(a.done) + _ = a.conn.Close() + }) +} + +// writePump serializes daemon→plugin frames (prompt/abort/welcome). +func (a *agentConn) writePump() { + for { + select { + case <-a.done: + return + case b := <-a.send: + _ = a.conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := a.conn.WriteMessage(websocket.TextMessage, b); err != nil { + a.drop() + return + } + } + } +} + +// Hub tracks live agent conns and web subscribers; it is the only component +// that mutates online state. +type Hub struct { + store *Store + + mu sync.Mutex + agents map[string]*agentConn + webs map[*webClient]struct{} + + // SpawnStatus lets the API push spawn job snapshots to web clients. + SpawnStatus func() []SpawnJob + + upgrader websocket.Upgrader +} + +func NewHub(store *Store) *Hub { + return &Hub{ + store: store, + agents: make(map[string]*agentConn), + webs: make(map[*webClient]struct{}), + upgrader: websocket.Upgrader{ + ReadBufferSize: 4096, + WriteBufferSize: 4096, + CheckOrigin: func(*http.Request) bool { return true }, + }, + } +} + +// IsOnline reports whether the session currently has a live agent WS. +func (h *Hub) IsOnline(sessionID string) bool { + h.mu.Lock() + defer h.mu.Unlock() + _, ok := h.agents[sessionID] + return ok +} + +// SessionsView merges persisted session rows with live online flags. +func (h *Hub) SessionsView() []SessionView { + rows, err := h.store.Sessions() + if err != nil { + log.Printf("hub: list sessions: %v", err) + return []SessionView{} + } + h.mu.Lock() + defer h.mu.Unlock() + out := make([]SessionView, 0, len(rows)) + for _, row := range rows { + v := SessionView{ + ID: row.Info.ID, + Name: row.Info.Name, + Cwd: row.Info.Cwd, + Model: row.Info.Model, + Provider: row.Info.Provider, + Agent: row.Info.Agent, + Repo: row.Info.Repo, + StartedAt: row.Info.StartedAt, + LastEventAt: row.LastEventAt, + } + _, v.Online = h.agents[row.ID] + out = append(out, v) + } + return out +} + +func (h *Hub) dropWeb(c *webClient) { + h.mu.Lock() + if _, ok := h.webs[c]; !ok { + h.mu.Unlock() + return + } + delete(h.webs, c) + h.mu.Unlock() + c.drop() + h.BroadcastSessionList() +} + +// BroadcastSessionList pushes a full session list to every web client. +func (h *Hub) BroadcastSessionList() { + payload, err := json.Marshal(map[string]any{ + "type": frameSessionList, + "sessions": h.SessionsView(), + }) + if err != nil { + return + } + h.mu.Lock() + clients := make([]*webClient, 0, len(h.webs)) + for c := range h.webs { + clients = append(clients, c) + } + h.mu.Unlock() + for _, c := range clients { + c.deliverControl(payload) + } +} + +// BroadcastSpawnStatus pushes the spawn job snapshot (if a provider is set). +func (h *Hub) BroadcastSpawnStatus() { + if h.SpawnStatus == nil { + return + } + jobs := h.SpawnStatus() + payload, err := json.Marshal(map[string]any{ + "type": frameSpawnStatus, + "jobs": jobs, + }) + if err != nil { + return + } + h.mu.Lock() + clients := make([]*webClient, 0, len(h.webs)) + for c := range h.webs { + clients = append(clients, c) + } + h.mu.Unlock() + for _, c := range clients { + c.deliverControl(payload) + } +} + +// publishEvent fans one agent frame out to web subscribers of that session. +func (h *Hub) publishEvent(sessionID string, seq int64, raw json.RawMessage) { + h.mu.Lock() + clients := make([]*webClient, 0, len(h.webs)) + for c := range h.webs { + clients = append(clients, c) + } + h.mu.Unlock() + ev := pendingEvent{sessionID: sessionID, seq: seq, raw: raw} + for _, c := range clients { + if c.subscription() == sessionID { + c.deliverEvent(ev) + } + } +} + +// ServeAgentWS handles GET /agent/ws (bearer auth already enforced by middleware). +func (h *Hub) ServeAgentWS(w http.ResponseWriter, r *http.Request) { + ws, err := h.upgrader.Upgrade(w, r, nil) + if err != nil { + return // Upgrade already wrote the HTTP error + } + ac := &agentConn{hub: h, conn: ws, send: make(chan []byte, agentSendQueue), done: make(chan struct{})} + go ac.writePump() + h.readAgentLoop(ac) +} + +func (h *Hub) readAgentLoop(ac *agentConn) { + defer ac.drop() + ac.conn.SetReadLimit(maxFrameSize) + registered := false + defer func() { + if registered { + h.unregister(ac) + } + }() + for { + _, data, err := ac.conn.ReadMessage() + if err != nil { + return + } + f, err := decodeFrame(data) + if err != nil { + log.Printf("hub: malformed frame from agent: %v", err) + continue // never kill the conn loop on a bad frame + } + switch f.typ { + case evHello: + registered = true + h.handleHello(ac, f) + case evSessionInfo: + h.handleSessionInfo(f) + case evBye: + return + default: + h.handleEvent(f) + } + } +} + +// handleHello persists the session snapshot, replaces any older conn, replies welcome. +func (h *Hub) handleHello(ac *agentConn, f frame) { + var fields struct { + Session SessionInfo `json:"session"` + } + if err := json.Unmarshal(f.raw, &fields); err != nil || fields.Session.ID == "" { + log.Printf("hub: hello without session payload from %s", f.sessionID) + return + } + sessionID := f.sessionID + if sessionID == "" { + sessionID = fields.Session.ID + } + fields.Session.ID = sessionID + ac.sessionID = sessionID + + h.mu.Lock() + if old, ok := h.agents[sessionID]; ok && old != ac { + old.drop() // reconnect: new conn replaces old + } + h.agents[sessionID] = ac + h.mu.Unlock() + + if err := h.store.UpsertSession(fields.Session); err != nil { + log.Printf("hub: persist session %s: %v", sessionID, err) + } + if err := h.store.SetOnline(sessionID, true); err != nil { + log.Printf("hub: mark online %s: %v", sessionID, err) + } + lastSeq, err := h.store.LastSeq(sessionID) + if err != nil { + log.Printf("hub: lastSeq %s: %v", sessionID, err) + lastSeq = 0 + } + welcome, err := json.Marshal(map[string]any{ + "v": daemonVersion, + "type": evWelcome, + "sessionId": sessionID, + "seq": 0, + "ts": time.Now().UnixMilli(), + "lastSeq": lastSeq, + }) + if err != nil { + return + } + select { + case ac.send <- welcome: + default: + ac.drop() + } + h.BroadcastSessionList() +} + +func (h *Hub) handleSessionInfo(f frame) { + var fields struct { + Session SessionInfo `json:"session"` + } + if err := json.Unmarshal(f.raw, &fields); err != nil || fields.Session.ID == "" { + return + } + sessionID := f.sessionID + if sessionID == "" { + sessionID = fields.Session.ID + } + fields.Session.ID = sessionID + if err := h.store.UpsertSession(fields.Session); err != nil { + log.Printf("hub: persist session_info %s: %v", sessionID, err) + } + h.BroadcastSessionList() +} + +// handleEvent persists (except message_update), updates bookkeeping, fans out live. +func (h *Hub) handleEvent(f frame) { + if f.sessionID == "" { + return + } + if f.typ != evMessageUpdate { + if err := h.store.AppendEvent(Event{ + SessionID: f.sessionID, + Seq: f.seq, + TS: f.ts, + Type: f.typ, + Payload: json.RawMessage(f.raw), + }); err != nil { + log.Printf("hub: persist event %s#%d: %v", f.sessionID, f.seq, err) + } + if err := h.store.TouchSession(f.sessionID, f.seq, f.ts); err != nil { + log.Printf("hub: touch session %s: %v", f.sessionID, err) + } + } + h.publishEvent(f.sessionID, f.seq, json.RawMessage(f.raw)) +} + +func (h *Hub) unregister(ac *agentConn) { + h.mu.Lock() + if cur, ok := h.agents[ac.sessionID]; ok && cur == ac { + delete(h.agents, ac.sessionID) + } + h.mu.Unlock() + if ac.sessionID != "" { + if err := h.store.SetOnline(ac.sessionID, false); err != nil { + log.Printf("hub: mark offline %s: %v", ac.sessionID, err) + } + } + h.BroadcastSessionList() +} + +// Prompt routes a prompt frame to the live plugin conn for sessionID. +func (h *Hub) Prompt(sessionID, message string) error { + return h.sendToAgent(sessionID, map[string]any{ + "v": daemonVersion, + "type": evPrompt, + "sessionId": sessionID, + "ts": time.Now().UnixMilli(), + "promptId": newUUID(), + "message": message, + }) +} + +// Abort routes an abort frame to the live plugin conn for sessionID. +func (h *Hub) Abort(sessionID string) error { + return h.sendToAgent(sessionID, map[string]any{ + "v": daemonVersion, + "type": evAbort, + "sessionId": sessionID, + "ts": time.Now().UnixMilli(), + }) +} + +func (h *Hub) sendToAgent(sessionID string, frameBody map[string]any) error { + b, err := json.Marshal(frameBody) + if err != nil { + return err + } + h.mu.Lock() + ac := h.agents[sessionID] + h.mu.Unlock() + if ac == nil { + return ErrOffline + } + select { + case ac.send <- b: + return nil + case <-ac.done: + return ErrOffline + case <-time.After(promptSendTimeout): + return ErrOffline + } +} + +// ServeWebWS handles GET /ws?token=... — token arrives as query param. +func (h *Hub) ServeWebWS(w http.ResponseWriter, r *http.Request) { + ws, err := h.upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + c := &webClient{hub: h, conn: ws, done: make(chan struct{})} + h.mu.Lock() + h.webs[c] = struct{}{} + h.mu.Unlock() + go c.writePump() + go c.readPump() + + payload, err := json.Marshal(map[string]any{ + "type": frameSessionList, + "sessions": h.SessionsView(), + }) + if err == nil { + c.deliverControl(payload) + } +} diff --git a/daemon/hub_test.go b/daemon/hub_test.go new file mode 100644 index 0000000..82ddf68 --- /dev/null +++ b/daemon/hub_test.go @@ -0,0 +1,332 @@ +package main + +// hub_test.go — agent handshake, replay, prompt routing, live fan-out. + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +const testToken string = "test-token" + +// newTestServer spins a Server over a temp store and returns both. +func newTestServer(t *testing.T) (*httptest.Server, *Store) { + t.Helper() + store := openTestStore(t) + daemonToken = testToken // tests run sequentially; package var set per-server + hub := NewHub(store) + srv := &Server{store: store, hub: hub, gitlab: NewGitLab(store, "https://gitlab.example")} + ts := httptest.NewServer(srv.Routes("")) + t.Cleanup(ts.Close) + return ts, store +} + +func dialAgent(t *testing.T, ts *httptest.Server) *websocket.Conn { + t.Helper() + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/agent/ws" + hdr := http.Header{"Authorization": []string{"Bearer " + testToken}} + ws, _, err := websocket.DefaultDialer.Dial(wsURL, hdr) + if err != nil { + t.Fatalf("dial agent ws: %v", err) + } + t.Cleanup(func() { _ = ws.Close() }) + return ws +} + +func readFrame(t *testing.T, ws *websocket.Conn) map[string]any { + t.Helper() + var m map[string]any + if err := ws.ReadJSON(&m); err != nil { + t.Fatalf("read frame: %v", err) + } + return m +} + +func helloFrame(sessionID string) map[string]any { + return map[string]any{ + "v": 1, "type": evHello, "sessionId": sessionID, "seq": 0, "ts": 1, + "session": map[string]any{ + "id": sessionID, "name": nil, "cwd": "/w", "model": "glm-5.3", + "provider": "zai-renaud", "agent": true, "repo": nil, "startedAt": 99, + }, + } +} + +func TestHubHandshakeWelcomeLastSeq(t *testing.T) { + ts, store := newTestServer(t) + ws := dialAgent(t, ts) + if err := ws.WriteJSON(helloFrame("s1")); err != nil { + t.Fatalf("send hello: %v", err) + } + if welcome := readFrame(t, ws); welcome["type"] != evWelcome || welcome["lastSeq"].(float64) != 0 { + t.Fatalf("welcome = %v", welcome) + } + + for _, typ := range []string{evMessageStart, evMessageEnd, evAgentSettled} { + if err := ws.WriteJSON(map[string]any{"v": 1, "type": typ, "sessionId": "s1", "seq": 1, "ts": 100}); err != nil { + t.Fatalf("send %s: %v", typ, err) + } + } + waitFor(t, 2*time.Second, func() bool { + seq, _ := store.LastSeq("s1") + return seq == 1 + }) + _ = ws.Close() + + ws2 := dialAgent(t, ts) + _ = ws2.WriteJSON(helloFrame("s1")) + welcome2 := readFrame(t, ws2) + if welcome2["lastSeq"].(float64) != 1 { + t.Fatalf("welcome lastSeq after reconnect = %v, want 1", welcome2["lastSeq"]) + } +} + +func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("condition not met before timeout") +} + +func TestHubReconnectReplacesOldConn(t *testing.T) { + ts, _ := newTestServer(t) + ws1 := dialAgent(t, ts) + _ = ws1.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws1) + + ws2 := dialAgent(t, ts) + _ = ws2.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws2) + + _ = ws1.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, _, err := ws1.ReadMessage(); err == nil { + t.Fatal("old agent conn should be closed after replacement") + } +} + +func TestHubPromptRoutingAndOffline409(t *testing.T) { + ts, _ := newTestServer(t) + + resp, err := http.Post(ts.URL+"/api/sessions/s1/prompt", + "application/json", strings.NewReader(`{"message":"hi"}`)) + if err != nil { + t.Fatalf("post prompt: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("unauthenticated prompt = %d, want 401", resp.StatusCode) + } + + postJSON := func(path, body string) int { + req, _ := http.NewRequest(http.MethodPost, ts.URL+path, strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("post %s: %v", path, err) + } + resp.Body.Close() + return resp.StatusCode + } + + if code := postJSON("/api/sessions/s1/prompt", `{"message":"hi"}`); code != http.StatusConflict { + t.Fatalf("offline prompt = %d, want 409", code) + } + + ws := dialAgent(t, ts) + _ = ws.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws) + + if code := postJSON("/api/sessions/s1/prompt", `{"message":"do it"}`); code != http.StatusOK { + t.Fatalf("online prompt = %d, want 200", code) + } + got := readFrame(t, ws) + if got["type"] != evPrompt || got["message"] != "do it" { + t.Fatalf("prompt frame = %v", got) + } + if _, ok := got["promptId"].(string); !ok { + t.Fatalf("prompt frame missing promptId: %v", got) + } + + if code := postJSON("/api/sessions/s1/abort", ""); code != http.StatusOK { + t.Fatalf("abort = %d, want 200", code) + } + abort := readFrame(t, ws) + if abort["type"] != evAbort { + t.Fatalf("abort frame = %v", abort) + } +} + +func TestHubMessageUpdateNotPersistedOthersAre(t *testing.T) { + ts, store := newTestServer(t) + ws := dialAgent(t, ts) + _ = ws.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws) + + frames := []map[string]any{ + {"v": 1, "type": evMessageUpdate, "sessionId": "s1", "seq": 1, "ts": 10, "delta": "chunk"}, + {"v": 1, "type": evMessageEnd, "sessionId": "s1", "seq": 2, "ts": 11, "message": map[string]any{"role": "assistant", "id": "m2", "text": "chunk"}}, + } + for _, f := range frames { + if err := ws.WriteJSON(f); err != nil { + t.Fatalf("send: %v", err) + } + } + waitFor(t, 2*time.Second, func() bool { + events, _ := store.EventsAfter("s1", 0, 100) + return len(events) == 1 + }) + events, _ := store.EventsAfter("s1", 0, 100) + if len(events) != 1 || events[0].Type != evMessageEnd || events[0].Seq != 2 { + t.Fatalf("persisted events = %+v, want only message_end seq 2", events) + } + if last, _ := store.LastSeq("s1"); last != 2 { + t.Fatalf("lastSeq = %d, want 2 (deltas unpersisted but seq advanced)", last) + } +} + +func TestHubMalformedFramesDoNotKillConn(t *testing.T) { + ts, _ := newTestServer(t) + ws := dialAgent(t, ts) + if err := ws.WriteMessage(websocket.TextMessage, []byte("not json")); err != nil { + t.Fatalf("send garbage: %v", err) + } + if err := ws.WriteJSON(map[string]any{"v": 1, "type": 42, "sessionId": "s1"}); err != nil { + t.Fatalf("send wrong-type: %v", err) + } + _ = ws.WriteJSON(helloFrame("s1")) + if welcome := readFrame(t, ws); welcome["type"] != evWelcome { + t.Fatalf("conn died after malformed frames: %v", welcome) + } +} + +func TestHubWebWSSubscribeReceivesLiveEvents(t *testing.T) { + ts, _ := newTestServer(t) + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/ws?token=" + testToken + web, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial web ws: %v", err) + } + t.Cleanup(func() { _ = web.Close() }) + + if first := readFrame(t, web); first["type"] != frameSessionList { + t.Fatalf("first web frame = %v", first) + } + + agent := dialAgent(t, ts) + _ = agent.WriteJSON(helloFrame("s1")) + _ = readFrame(t, agent) + + // session_list broadcast on connect arrives before the event stream. + deadline := time.Now().Add(2 * time.Second) + _ = web.SetReadDeadline(deadline) + for { + if m := readFrame(t, web); m["type"] == frameSessionList { + break + } + } + _ = web.WriteJSON(map[string]any{"type": frameSubscribe, "sessionId": "s1"}) + + // The subscription registers server-side asynchronously; an event sent too + // early would be dropped (per protocol the UI refetches from REST). Resend + // until a batch arrives. A reader goroutine avoids deadline-based reads + // (gorilla conns fail permanently after a read timeout). + frames := make(chan map[string]any, 16) + go func() { + defer close(frames) + for { + var m map[string]any + if err := web.ReadJSON(&m); err != nil { + return + } + frames <- m + } + }() + start := time.Now() + seq := int64(5) + var got map[string]any + for got == nil { + seq++ + _ = agent.WriteJSON(map[string]any{"v": 1, "type": evMessageUpdate, "sessionId": "s1", "seq": seq, "ts": 50, "delta": "hi"}) + select { + case m, ok := <-frames: + if !ok { + t.Fatal("web conn closed before events frame") + } + if m["type"] == frameEvents { + got = m + } + case <-time.After(200 * time.Millisecond): + } + if time.Since(start) > 3*time.Second { + t.Fatal("no events frame delivered after subscribe") + } + } + evList := got["events"].([]any) + if len(evList) == 0 { + t.Fatalf("events batch empty: %v", got) + } + first := evList[0].(map[string]any) + if got["sessionId"] != "s1" || got["after"].(float64) != first["seq"].(float64)-1 { + t.Fatalf("events frame = %v", got) + } + for _, e := range evList { + if e.(map[string]any)["type"] != evMessageUpdate { + t.Fatalf("unexpected event in batch: %v", e) + } + } +} + +func TestHubWebWSBadTokenRejected(t *testing.T) { + ts, _ := newTestServer(t) + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/ws?token=wrong" + if web, _, err := websocket.DefaultDialer.Dial(wsURL, nil); err == nil { + _ = web.Close() + t.Fatal("web ws with bad token should be rejected before upgrade") + } +} + +func TestHubSessionsViewOnlineFlag(t *testing.T) { + ts, _ := newTestServer(t) + ws := dialAgent(t, ts) + _ = ws.WriteJSON(helloFrame("s1")) + _ = readFrame(t, ws) + + fetchSessions := func() []map[string]any { + req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/sessions", nil) + req.Header.Set("Authorization", "Bearer "+testToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("get sessions: %v", err) + } + defer resp.Body.Close() + var sessions []map[string]any + if err := json.NewDecoder(resp.Body).Decode(&sessions); err != nil { + t.Fatalf("decode: %v", err) + } + return sessions + } + + sessions := fetchSessions() + if len(sessions) != 1 || sessions[0]["id"] != "s1" { + t.Fatalf("sessions = %v", sessions) + } + if sessions[0]["online"] != true { + t.Fatalf("online flag = %v, want true", sessions[0]["online"]) + } + + _ = ws.Close() + waitFor(t, 2*time.Second, func() bool { + return fetchSessions()[0]["online"] == false + }) +} diff --git a/daemon/main.go b/daemon/main.go new file mode 100644 index 0000000..c4a4b89 --- /dev/null +++ b/daemon/main.go @@ -0,0 +1,86 @@ +package main + +// main.go — lvmh daemon entrypoint: env wiring, flags, HTTP server. + +import ( + "context" + "flag" + "log" + "net/http" + "os" + "os/signal" + "path/filepath" + "syscall" + "time" +) + +// Server and env wiring constants. +const ( + envDB string = "LVMH_DB" + defaultDBPath string = "/data/lvmh.db" + envGitLabBaseURL string = "GITLAB_BASE_URL" + defaultGitLabBase string = "https://git.westphal.fr" + defaultListenAddr string = ":8686" + shutdownGrace time.Duration = 10 * time.Second + webDistDefault string = "/app/web-dist" +) + +func main() { + var ( + addr = flag.String("addr", defaultListenAddr, "listen address") + dbPath = flag.String("db", envOr(envDB, defaultDBPath), "sqlite database path") + webdist = flag.String("webdist", webDistDefault, "serve web UI from this directory instead of the embedded build") + ) + flag.Parse() + // Fall back to the embedded UI when the override directory is absent. + if st, err := os.Stat(*webdist); err != nil || !st.IsDir() { + *webdist = "" + } + + daemonToken = os.Getenv(envToken) + if daemonToken == "" { + log.Fatalf("LVMH_TOKEN must be set") + } + + if dir := filepath.Dir(*dbPath); dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o755); err != nil { + log.Fatalf("create db dir: %v", err) + } + } + store, err := OpenStore(*dbPath) + if err != nil { + log.Fatalf("open store: %v", err) + } + defer store.Close() + + hub := NewHub(store) + gitlab := NewGitLab(store, envOr(envGitLabBaseURL, defaultGitLabBase)) + spawner, err := NewSpawner(store, hub, envOr(envGitLabBaseURL, defaultGitLabBase)) + if err != nil { + log.Fatalf("docker client: %v", err) + } + + srv := NewServer(store, hub, spawner, gitlab) + httpServer := &http.Server{ + Addr: *addr, + Handler: srv.Routes(*webdist), + ReadHeaderTimeout: 10 * time.Second, + } + + go func() { + log.Printf("lvmh daemon listening on %s (db %s)", *addr, *dbPath) + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("listen: %v", err) + } + }() + + stop := make(chan os.Signal, 1) + signal.Notify(stop, os.Interrupt, syscall.SIGTERM) + <-stop + log.Printf("shutting down") + ctx, cancel := context.WithTimeout(context.Background(), shutdownGrace) + defer cancel() + if err := httpServer.Shutdown(ctx); err != nil { + log.Printf("shutdown: %v", err) + } +} diff --git a/daemon/store.go b/daemon/store.go new file mode 100644 index 0000000..486d6f5 --- /dev/null +++ b/daemon/store.go @@ -0,0 +1,239 @@ +package main + +// store.go — SQLite persistence: sessions, events, settings, containers. + +import ( + "database/sql" + "encoding/json" + "fmt" + + _ "modernc.org/sqlite" +) + +// SessionInfo mirrors the hello/session_info payload snapshot in PROTOCOL.md. +type SessionInfo struct { + ID string `json:"id"` + Name *string `json:"name"` + Cwd string `json:"cwd"` + Model string `json:"model"` + Provider string `json:"provider"` + Agent bool `json:"agent"` + Repo *string `json:"repo"` + StartedAt int64 `json:"startedAt"` +} + +// SessionRow is a persisted session with its bookkeeping columns. +type SessionRow struct { + ID string `json:"id"` + Info SessionInfo `json:"-"` + LastSeq int64 `json:"-"` + LastEventAt int64 `json:"lastEventAt"` + OnlineDB bool `json:"-"` +} + +// Event is a persisted agent event; Payload is the raw JSON payload object. +type Event struct { + SessionID string + Seq int64 + TS int64 + Type string + Payload json.RawMessage +} + +// ContainerRow tracks a daemon-spawned worker container per session. +type ContainerRow struct { + SessionID string + ContainerID string + Repo string +} + +const dsnParams string = "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)" + +const schema string = ` +CREATE TABLE IF NOT EXISTS events( + sessionId TEXT NOT NULL, seq INTEGER NOT NULL, ts INTEGER NOT NULL, + type TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(sessionId, seq)); +CREATE TABLE IF NOT EXISTS sessions( + id TEXT PRIMARY KEY, info TEXT NOT NULL, + lastSeq INTEGER NOT NULL DEFAULT 0, lastEventAt INTEGER, online INTEGER DEFAULT 0); +CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY, value TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS containers( + sessionId TEXT PRIMARY KEY, containerId TEXT NOT NULL, repo TEXT NOT NULL); +CREATE INDEX IF NOT EXISTS idx_events_session ON events(sessionId, seq); +` + +// OpenStore opens (creating if needed) the SQLite database at path with WAL. +func OpenStore(path string) (*Store, error) { + db, err := sql.Open("sqlite", path+dsnParams) + if err != nil { + return nil, fmt.Errorf("open sqlite: %w", err) + } + // modernc sqlite serializes writes; a single conn avoids SQLITE_BUSY churn. + db.SetMaxOpenConns(1) + if _, err := db.Exec(schema); err != nil { + db.Close() + return nil, fmt.Errorf("apply schema: %w", err) + } + return &Store{db: db}, nil +} + +// Store wraps the SQLite handle. +type Store struct { + db *sql.DB +} + +// Close closes the database. +func (s *Store) Close() error { return s.db.Close() } + +// AppendEvent persists one event keyed (sessionId, seq). Duplicates are ignored. +func (s *Store) AppendEvent(e Event) error { + payload := e.Payload + if len(payload) == 0 { + payload = json.RawMessage("{}") + } + _, err := s.db.Exec( + `INSERT OR IGNORE INTO events(sessionId, seq, ts, type, payload) VALUES(?,?,?,?,?)`, + e.SessionID, e.Seq, e.TS, e.Type, string(payload)) + return err +} + +// LastSeq returns the highest persisted seq for the session, 0 if none. +func (s *Store) LastSeq(sessionID string) (int64, error) { + var seq sql.NullInt64 + err := s.db.QueryRow(`SELECT MAX(seq) FROM events WHERE sessionId=?`, sessionID).Scan(&seq) + if err != nil { + return 0, err + } + return seq.Int64, nil +} + +// EventsAfter returns persisted events for the session with seq > after, +// ascending, at most limit rows. +func (s *Store) EventsAfter(sessionID string, after int64, limit int) ([]Event, error) { + rows, err := s.db.Query( + `SELECT sessionId, seq, ts, type, payload FROM events + WHERE sessionId=? AND seq>? ORDER BY seq LIMIT ?`, + sessionID, after, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Event + for rows.Next() { + var e Event + var payload string + if err := rows.Scan(&e.SessionID, &e.Seq, &e.TS, &e.Type, &payload); err != nil { + return nil, err + } + e.Payload = json.RawMessage(payload) + out = append(out, e) + } + return out, rows.Err() +} + +// UpsertSession inserts or refreshes the info snapshot for a session. +func (s *Store) UpsertSession(info SessionInfo) error { + blob, err := json.Marshal(info) + if err != nil { + return err + } + _, err = s.db.Exec( + `INSERT INTO sessions(id, info, lastSeq, lastEventAt, online) VALUES(?,?,0,NULL,0) + ON CONFLICT(id) DO UPDATE SET info=excluded.info`, + info.ID, string(blob)) + return err +} + +// TouchSession advances lastSeq (monotonic) and lastEventAt. +func (s *Store) TouchSession(sessionID string, lastSeq, lastEventAt int64) error { + _, err := s.db.Exec( + `UPDATE sessions SET lastSeq=MAX(lastSeq,?), lastEventAt=? WHERE id=?`, + lastSeq, lastEventAt, sessionID) + return err +} + +// SetOnline records the persisted online flag (memory hub is authoritative). +func (s *Store) SetOnline(sessionID string, online bool) error { + _, err := s.db.Exec(`UPDATE sessions SET online=? WHERE id=?`, online, sessionID) + return err +} + +// Sessions returns every known session row ordered oldest-first. +func (s *Store) Sessions() ([]SessionRow, error) { + rows, err := s.db.Query(`SELECT id, info, lastSeq, lastEventAt, online FROM sessions ORDER BY id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []SessionRow + for rows.Next() { + var row SessionRow + var infoBlob string + var lastEventAt sql.NullInt64 + var online bool + if err := rows.Scan(&row.ID, &infoBlob, &row.LastSeq, &lastEventAt, &online); err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(infoBlob), &row.Info); err != nil { + return nil, fmt.Errorf("decode session info %s: %w", row.ID, err) + } + row.LastEventAt = lastEventAt.Int64 + row.OnlineDB = online + out = append(out, row) + } + return out, rows.Err() +} + +// GetSetting returns a settings value and whether it exists. +func (s *Store) GetSetting(key string) (string, bool, error) { + var val string + err := s.db.QueryRow(`SELECT value FROM settings WHERE key=?`, key).Scan(&val) + if err == sql.ErrNoRows { + return "", false, nil + } + if err != nil { + return "", false, err + } + return val, true, nil +} + +// SetSetting upserts a settings value. +func (s *Store) SetSetting(key, value string) error { + _, err := s.db.Exec(`INSERT INTO settings(key, value) VALUES(?,?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value`, key, value) + return err +} + +// DeleteSetting removes a settings key. +func (s *Store) DeleteSetting(key string) error { + _, err := s.db.Exec(`DELETE FROM settings WHERE key=?`, key) + return err +} + +// UpsertContainer records the container created for a session. +func (s *Store) UpsertContainer(sessionID, containerID, repo string) error { + _, err := s.db.Exec(`INSERT INTO containers(sessionId, containerId, repo) VALUES(?,?,?) + ON CONFLICT(sessionId) DO UPDATE SET containerId=excluded.containerId, repo=excluded.repo`, + sessionID, containerID, repo) + return err +} + +// GetContainer returns the container row for a session, if any. +func (s *Store) GetContainer(sessionID string) (ContainerRow, bool, error) { + var row ContainerRow + err := s.db.QueryRow(`SELECT sessionId, containerId, repo FROM containers WHERE sessionId=?`, + sessionID).Scan(&row.SessionID, &row.ContainerID, &row.Repo) + if err == sql.ErrNoRows { + return row, false, nil + } + if err != nil { + return row, false, err + } + return row, true, nil +} + +// DeleteContainer forgets the container row for a session. +func (s *Store) DeleteContainer(sessionID string) error { + _, err := s.db.Exec(`DELETE FROM containers WHERE sessionId=?`, sessionID) + return err +} diff --git a/daemon/store_test.go b/daemon/store_test.go new file mode 100644 index 0000000..22c05c4 --- /dev/null +++ b/daemon/store_test.go @@ -0,0 +1,134 @@ +package main + +// store_test.go — persistence: events replay, sessions, settings, containers. + +import ( + "path/filepath" + "testing" +) + +func openTestStore(t *testing.T) *Store { + t.Helper() + store, err := OpenStore(filepath.Join(t.TempDir(), "lvmh.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + return store +} + +func TestStoreEventPersistAndReplayAfterSeq(t *testing.T) { + store := openTestStore(t) + events := []Event{ + {SessionID: "s1", Seq: 1, TS: 1000, Type: evMessageStart, Payload: []byte(`{"message":{"role":"user","id":"m1"}}`)}, + {SessionID: "s1", Seq: 2, TS: 1001, Type: evMessageEnd, Payload: []byte(`{"message":{"role":"user","id":"m1"}}`)}, + {SessionID: "s1", Seq: 3, TS: 1002, Type: evAgentSettled, Payload: []byte(`{}`)}, + } + for _, e := range events { + if err := store.AppendEvent(e); err != nil { + t.Fatalf("append seq %d: %v", e.Seq, err) + } + } + + if last, _ := store.LastSeq("s1"); last != 3 { + t.Fatalf("lastSeq = %d, want 3", last) + } + + all, err := store.EventsAfter("s1", 0, 100) + if err != nil { + t.Fatalf("events after 0: %v", err) + } + if len(all) != 3 { + t.Fatalf("after=0 returned %d events, want 3", len(all)) + } + for i, e := range all { + if e.Seq != int64(i+1) { + t.Fatalf("event %d has seq %d, want ascending from 1", i, e.Seq) + } + } + + tail, err := store.EventsAfter("s1", 2, 100) + if err != nil { + t.Fatalf("events after 2: %v", err) + } + if len(tail) != 1 || tail[0].Seq != 3 { + t.Fatalf("after=2 returned %v, want only seq 3", tail) + } + + limited, _ := store.EventsAfter("s1", 0, 2) + if len(limited) != 2 { + t.Fatalf("limit=2 returned %d events", len(limited)) + } + + // Duplicate (sessionId, seq) is ignored — replay idempotence. + if err := store.AppendEvent(events[0]); err != nil { + t.Fatalf("duplicate append: %v", err) + } + dup, _ := store.EventsAfter("s1", 0, 100) + if len(dup) != 3 { + t.Fatalf("duplicate append created extra row: %d", len(dup)) + } +} + +func TestStoreSessionsOnlineAndTouch(t *testing.T) { + store := openTestStore(t) + name := "my session" + info := SessionInfo{ID: "s1", Name: &name, Cwd: "/w", Model: "glm-5.3", Provider: "zai-renaud", Agent: true, Repo: nil, StartedAt: 42} + if err := store.UpsertSession(info); err != nil { + t.Fatalf("upsert: %v", err) + } + if err := store.TouchSession("s1", 7, 4242); err != nil { + t.Fatalf("touch: %v", err) + } + if err := store.TouchSession("s1", 3, 9999); err != nil { // lastSeq monotonic + t.Fatalf("touch lower: %v", err) + } + if err := store.SetOnline("s1", true); err != nil { + t.Fatalf("online: %v", err) + } + + rows, err := store.Sessions() + if err != nil || len(rows) != 1 { + t.Fatalf("sessions: %v %d", err, len(rows)) + } + row := rows[0] + if row.LastSeq != 7 || row.LastEventAt != 9999 || !row.OnlineDB { + t.Fatalf("row = %+v, want lastSeq 7, lastEventAt 9999, online", row) + } + if row.Info.Name == nil || *row.Info.Name != name { + t.Fatalf("name not roundtripped: %+v", row.Info) + } + if last, _ := store.LastSeq("nope"); last != 0 { + t.Fatalf("missing session lastSeq = %d, want 0", last) + } +} + +func TestStoreSettingsAndContainers(t *testing.T) { + store := openTestStore(t) + if _, ok, _ := store.GetSetting(settingGitLabToken); ok { + t.Fatal("unset setting should not exist") + } + if err := store.SetSetting(settingGitLabToken, "pat"); err != nil { + t.Fatalf("set: %v", err) + } + if v, ok, _ := store.GetSetting(settingGitLabToken); !ok || v != "pat" { + t.Fatalf("get = %q %v", v, ok) + } + if err := store.DeleteSetting(settingGitLabToken); err != nil { + t.Fatalf("delete: %v", err) + } + + if err := store.UpsertContainer("s1", "cid1", "group/proj"); err != nil { + t.Fatalf("upsert container: %v", err) + } + row, ok, err := store.GetContainer("s1") + if err != nil || !ok || row.ContainerID != "cid1" || row.Repo != "group/proj" { + t.Fatalf("get container = %+v %v %v", row, ok, err) + } + if err := store.DeleteContainer("s1"); err != nil { + t.Fatalf("delete container: %v", err) + } + if _, ok, _ := store.GetContainer("s1"); ok { + t.Fatal("container row should be gone") + } +} diff --git a/daemon/webdist/.gitkeep b/daemon/webdist/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/daemon/webdist/index.html b/daemon/webdist/index.html new file mode 100644 index 0000000..a66190d --- /dev/null +++ b/daemon/webdist/index.html @@ -0,0 +1,24 @@ + + + + + + lvmh + + + +
+

lvmh daemon

+

+ Web UI not bundled in this build. Build web/ and pass + --webdist /app/web-dist (or copy web/dist over + daemon/webdist/ before building the image). +

+

API: /api/sessions · agent WS: /agent/ws · web WS: /ws

+
+ +