lvmh/daemon/api.go

package main

// api.go — REST /api/* handlers, bearer auth, static web serving.

import (
	"crypto/subtle"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
)

// Route names, event query defaults and limits.
const (
	envToken string = "LVMH_TOKEN"

	envModelsFile     string = "LVMH_MODELS_FILE"
	defaultModelsFile string = "/app/build/docker/worker-models.json"

	defaultEventsAfter int64  = 0
	defaultEventsLimit int    = 1000
	maxEventsLimit     int    = 10000
	maxBodyBytes       int64  = 1 << 20
	webIndexFallback   string = "index.html"

	reposPathPrefix    string = "/api/repos/"
	reposPrepareSuffix string = "/prepare"
	// opsPreparePrompt is sent to the ops control session by
	// POST /api/repos/<repo>/prepare.
	opsPreparePrompt string = "prepare %s: clone, build a worker image, register it"
	reposPathSuffix  string = "/image"
)

// imageNameRe pins registered images to the lvmh-worker- namespace so agents
// can only point spawns at images we built.
var imageNameRe = regexp.MustCompile(`^lvmh-worker-[a-z0-9][a-z0-9.-]*:?[a-zA-Z0-9._-]*$`)

//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.Handle("GET /agent/ws", s.bearerAuth(http.HandlerFunc(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("GET /api/sessions/{id}/stats", s.handleSessionStats)
	api.HandleFunc("GET /api/stats", s.handleStats)
	api.HandleFunc("POST /api/sessions/{id}/prompt", s.handlePrompt)
	api.HandleFunc("POST /api/sessions/{id}/abort", s.handleAbort)
	api.HandleFunc("POST /api/sessions/{id}/model", s.handleSetModel)
	api.HandleFunc("PATCH /api/sessions/{id}", s.handleRenameSession)
	api.HandleFunc("DELETE /api/sessions/{id}/container", s.handleDeleteContainer)
	api.HandleFunc("GET /api/model-catalog", s.handleModelCatalog)
	api.HandleFunc("POST /api/spawn", s.handleSpawn)
	api.HandleFunc("GET /api/spawn/status", s.handleSpawnStatus)
	api.HandleFunc("GET /api/repos", s.handleRepoImages)
	api.HandleFunc("POST /api/repos/", s.handlePrepareRepo)
	api.HandleFunc("POST /api/pi-config/resync", s.handlePiConfigResync)
	// repo paths contain "/" (group/project), which a single {repo} wildcard
	// segment cannot match — subtree routes with manual path parsing instead.
	api.HandleFunc("PUT /api/repos/", s.handleSetRepoImage)
	api.HandleFunc("DELETE /api/repos/", s.handleDeleteRepoImage)
	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
	}
	var (
		events []Event
		err    error
	)
	switch {
	case r.URL.Query().Get("latest") != "":
		if r.URL.Query().Get("latest") != "1" {
			writeError(w, http.StatusBadRequest, "invalid latest")
			return
		}
		events, err = s.store.EventsLatest(id, limit)
	case r.URL.Query().Get("before") != "":
		before, perr := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64)
		if perr != nil || before < 0 {
			writeError(w, http.StatusBadRequest, "invalid before")
			return
		}
		events, err = s.store.EventsBefore(id, before, limit)
	default:
		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) handleSetModel(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	var body struct {
		Provider string `json:"provider"`
		ModelID  string `json:"modelId"`
	}
	if !decodeBody(w, r, &body) {
		return
	}
	if strings.TrimSpace(body.ModelID) == "" || strings.TrimSpace(body.Provider) == "" {
		writeError(w, http.StatusBadRequest, "provider and modelId required")
		return
	}
	if err := s.hub.SetModel(id, body.Provider, body.ModelID); 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) handleRenameSession(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	var body struct {
		Name string `json:"name"`
	}
	if !decodeBody(w, r, &body) {
		return
	}
	if strings.TrimSpace(body.Name) == "" {
		writeError(w, http.StatusBadRequest, "name required")
		return
	}
	found, err := s.store.SetSessionName(id, body.Name)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	if !found {
		writeError(w, http.StatusNotFound, "unknown session")
		return
	}
	// best-effort mirror into the live session; offline renames persist
	// daemon-side only (documented edge in PROTOCOL.md)
	if err := s.hub.Rename(id, body.Name); err != nil && !errors.Is(err, ErrOffline) {
		log.Printf("api: rename frame %s: %v", id, err)
	}
	s.hub.BroadcastSessionList()
	writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}

func (s *Server) handleDeleteContainer(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	var err error
	if id == opsSessionID {
		// the ops container has no containers-table row; remove by name
		err = s.spawn.RemoveOps(r.Context())
	} else {
		err = s.spawn.RemoveSession(r.Context(), id)
	}
	if 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
	}
	if body.Branch != "" && !branchRe.MatchString(body.Branch) {
		writeError(w, http.StatusBadRequest, "invalid branch name")
		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())
}

// repoFromImagePath extracts the repo path from /api/repos/<repo>/image.
func repoFromImagePath(path string) (string, bool) {
	rest, ok := strings.CutSuffix(strings.TrimPrefix(path, reposPathPrefix), reposPathSuffix)
	if !ok || !validRepoPath(rest) {
		return "", false
	}
	return rest, true
}

func (s *Server) handleRepoImages(w http.ResponseWriter, r *http.Request) {
	rows, err := s.store.ListRepoImages()
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	if rows == nil {
		rows = []RepoImageRow{}
	}
	for i := range rows {
		built, err := s.spawn.imageRefExists(r.Context(), rows[i].Image)
		if err != nil {
			writeError(w, http.StatusInternalServerError, "docker: "+err.Error())
			return
		}
		rows[i].Built = built
	}
	writeJSON(w, http.StatusOK, rows)
}

func (s *Server) handleSetRepoImage(w http.ResponseWriter, r *http.Request) {
	repo, ok := repoFromImagePath(r.URL.Path)
	if !ok {
		writeError(w, http.StatusBadRequest, "repo must look like group/project")
		return
	}
	var body struct {
		Image string `json:"image"`
	}
	if !decodeBody(w, r, &body) {
		return
	}
	if !imageNameRe.MatchString(body.Image) {
		writeError(w, http.StatusBadRequest, "invalid image name")
		return
	}
	if err := s.store.SetRepoImage(repo, body.Image); err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}

func (s *Server) handleDeleteRepoImage(w http.ResponseWriter, r *http.Request) {
	repo, ok := repoFromImagePath(r.URL.Path)
	if !ok {
		writeError(w, http.StatusBadRequest, "repo must look like group/project")
		return
	}
	if err := s.store.DeleteRepoImage(repo); err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}

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)
}

// ModelCatalogItem is one selectable model in GET /api/model-catalog.
type ModelCatalogItem struct {
	Provider string `json:"provider"`
	ID       string `json:"id"`
	Name     string `json:"name"`
}

// parseModelCatalog extracts provider/id/name entries from a pi models.json
// file body, grouped per provider (file order preserved within a provider).
func parseModelCatalog(data []byte) []ModelCatalogItem {
	var doc struct {
		Providers map[string]struct {
			Models []struct {
				ID   string `json:"id"`
				Name string `json:"name"`
			} `json:"models"`
		} `json:"providers"`
	}
	if err := json.Unmarshal(data, &doc); err != nil {
		return nil
	}
	providers := make([]string, 0, len(doc.Providers))
	for p := range doc.Providers {
		providers = append(providers, p)
	}
	sort.Strings(providers)
	out := []ModelCatalogItem{}
	for _, p := range providers {
		for _, m := range doc.Providers[p].Models {
			if m.ID == "" {
				continue
			}
			name := m.Name
			if name == "" {
				name = m.ID
			}
			out = append(out, ModelCatalogItem{Provider: p, ID: m.ID, Name: name})
		}
	}
	return out
}

func (s *Server) handleModelCatalog(w http.ResponseWriter, r *http.Request) {
	data, err := os.ReadFile(envOr(envModelsFile, defaultModelsFile))
	if err != nil {
		writeError(w, http.StatusInternalServerError, "model catalog unavailable: "+err.Error())
		return
	}
	catalog := parseModelCatalog(data)
	if catalog == nil {
		writeError(w, http.StatusInternalServerError, "model catalog unreadable")
		return
	}
	writeJSON(w, http.StatusOK, catalog)
}

// 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)
	})
}

// daemonToken is set by main from LVMH_TOKEN (single process, read-only).
var daemonToken string

// repoFromPreparePath extracts the repo path from /api/repos/<repo>/prepare.
func repoFromPreparePath(path string) (string, bool) {
	return repoFromSubPath(path, reposPrepareSuffix)
}

func (s *Server) handlePrepareRepo(w http.ResponseWriter, r *http.Request) {
	repo, ok := repoFromPreparePath(r.URL.Path)
	if !ok {
		writeError(w, http.StatusBadRequest, "repo must look like group/project")
		return
	}
	if err := s.hub.Prompt(opsSessionID, fmt.Sprintf(opsPreparePrompt, repo)); err != nil {
		if errors.Is(err, ErrOffline) {
			writeError(w, http.StatusConflict, "ops session offline")
			return
		}
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}

func repoFromSubPath(path, suffix string) (string, bool) {
	rest, ok := strings.CutSuffix(strings.TrimPrefix(path, reposPathPrefix), suffix)
	if !ok || !validRepoPath(rest) {
		return "", false
	}
	return rest, true
}

func (s *Server) handleSessionStats(w http.ResponseWriter, r *http.Request) {
	stats, err := s.store.SessionStats(r.PathValue("id"))
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, stats)
}

func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
	totals, err := s.store.StatsTotals()
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, statsResponse{StatsTotals: totals, OnlineCount: s.hub.OnlineCount()})
}

// statsResponse is the /api/stats shape: daemon-wide totals plus hub liveness.
type statsResponse struct {
	StatsTotals
	OnlineCount int `json:"onlineCount"`
}