package-updater/internal/master/web/server.go

// Package web serves the master's htmx UI.
package web

import (
	"context"
	"embed"
	"errors"
	"fmt"
	"html/template"
	"log/slog"
	"net/http"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/shcizo/package-updater/internal/agenttoken"
	"github.com/shcizo/package-updater/internal/master/events"
	"github.com/shcizo/package-updater/internal/master/model"
	"github.com/shcizo/package-updater/internal/master/webauth"
)

//go:embed templates/*.html static/*
var assets embed.FS

// serviceJobsLimit is how many jobs the service page lists.
const serviceJobsLimit = 20

var agentNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`)

// Store is the storage the UI reads and writes. *store.Store satisfies it.
type Store interface {
	ListServices(ctx context.Context) ([]model.Service, error)
	GetService(ctx context.Context, id int64) (model.Service, error)
	GetJob(ctx context.Context, id int64) (model.Job, error)
	ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error)
	ListAgents(ctx context.Context) ([]model.Agent, error)
	CreateAgent(ctx context.Context, name, tokenHash string) (model.Agent, error)
	RevokeAgent(ctx context.Context, id int64, at time.Time) error
}

// Hub reports and controls live agent connections. *hub.Hub satisfies it.
type Hub interface {
	Online(agentID int64) bool
	Disconnect(agentID int64)
}

// Jobs starts updates and explains why a service cannot be updated.
// *jobs.Service satisfies it.
type Jobs interface {
	Updatable(svc model.Service) error
	Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error)
}

// Broker delivers change signals to open streams and accepts signals from
// UI actions. *events.Broker satisfies it.
type Broker interface {
	Subscribe() (<-chan events.Topic, func())
	Publish(t events.Topic)
}

// Config holds values shown in the "add agent" snippet.
type Config struct {
	AgentURL   string // what agents dial, e.g. wss://10.0.0.5:8090/agent
	MasterKey  string // the master's fingerprint
	AgentImage string
}

// Server renders the UI.
type Server struct {
	store  Store
	hub    Hub
	jobs   Jobs
	broker Broker
	cfg    Config
	log    *slog.Logger
	now    func() time.Time
	pages  map[string]*template.Template
}

// New parses the templates. Each page is its own template set so every page
// can define its own "content" block.
func New(store Store, hub Hub, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) (*Server, error) {
	funcs := template.FuncMap{"shortRef": shortRef, "duration": jobDuration}
	pages := map[string]*template.Template{}
	for _, page := range []string{"overview.html", "agents.html", "agent_created.html", "service.html", "jobs.html", "job.html"} {
		t, err := template.New(page).Funcs(funcs).ParseFS(assets, "templates/layout.html", "templates/"+page)
		if err != nil {
			return nil, fmt.Errorf("parse %s: %w", page, err)
		}
		pages[page] = t
	}
	return &Server{store: store, hub: hub, jobs: jobs, broker: broker, cfg: cfg, log: log, now: time.Now, pages: pages}, nil
}

// Register adds the UI routes to mux. protect wraps every route except
// static assets, normally with webauth's Require.
func (s *Server) Register(mux *http.ServeMux, protect func(http.Handler) http.Handler) {
	mux.Handle("GET /static/", http.FileServerFS(assets))
	route := func(pattern string, h http.HandlerFunc) { mux.Handle(pattern, protect(h)) }
	route("GET /{$}", s.overview)
	route("GET /fragments/services", s.servicesFragment)
	route("GET /services/{id}", s.servicePage)
	route("GET /fragments/services/{id}/jobs", s.serviceJobsFragment)
	route("POST /services/{id}/update", s.updateService)
	route("GET /jobs", s.jobsPage)
	route("GET /fragments/jobs", s.jobsFragment)
	route("GET /jobs/{id}", s.jobPage)
	route("GET /fragments/jobs/{id}", s.jobFragment)
	route("GET /agents", s.agents)
	route("GET /fragments/agents", s.agentsFragment)
	route("POST /agents", s.createAgent)
	route("POST /agents/{id}/revoke", s.revokeAgent)
	route("GET /events", s.events)
}

type pageData struct {
	Title   string
	CSRF    string
	Groups  []serviceGroup
	Agents  []agentRow
	Error   string
	Created model.Agent
	Snippet string

	Service     model.Service
	ConfigFiles []string
	Reason      string // why Service cannot be updated; "" when it can
	Jobs        []model.Job
	Job         model.Job
}

func (s *Server) base(r *http.Request, title string) pageData {
	sess, _ := webauth.SessionFrom(r.Context())
	return pageData{Title: title, CSRF: sess.CSRFToken}
}

func (s *Server) render(w http.ResponseWriter, status int, page, block string, data pageData) {
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	w.WriteHeader(status)
	if err := s.pages[page].ExecuteTemplate(w, block, data); err != nil {
		s.log.Error("render_failed", "page", page, "block", block, "error", err)
	}
}

func (s *Server) serviceData(r *http.Request) (pageData, error) {
	data := s.base(r, "Services")
	services, err := s.store.ListServices(r.Context())
	if err != nil {
		return data, err
	}
	data.Groups = groupServices(services)
	return data, nil
}

func (s *Server) overview(w http.ResponseWriter, r *http.Request) {
	data, err := s.serviceData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	s.render(w, http.StatusOK, "overview.html", "layout", data)
}

func (s *Server) servicesFragment(w http.ResponseWriter, r *http.Request) {
	data, err := s.serviceData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	s.render(w, http.StatusOK, "overview.html", "services", data)
}

// pathID parses the {id} path value. On failure it has already answered 400.
func pathID(w http.ResponseWriter, r *http.Request) (int64, bool) {
	id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
	if err != nil {
		http.Error(w, "invalid id", http.StatusBadRequest)
		return 0, false
	}
	return id, true
}

// failLookup answers 404 for model.ErrNotFound and 500 otherwise.
func (s *Server) failLookup(w http.ResponseWriter, err error) {
	if errors.Is(err, model.ErrNotFound) {
		http.Error(w, "not found", http.StatusNotFound)
		return
	}
	s.fail(w, err)
}

func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) {
	svc, err := s.store.GetService(r.Context(), id)
	if err != nil {
		return pageData{}, err
	}
	data := s.base(r, svc.DisplayName())
	data.Service = svc
	if svc.ConfigFiles != "" {
		data.ConfigFiles = strings.Split(svc.ConfigFiles, ",")
	}
	var nu *model.NotUpdatableError
	if err := s.jobs.Updatable(svc); errors.As(err, &nu) {
		data.Reason = nu.Reason
	} else if err != nil {
		return data, err
	}
	data.Jobs, err = s.store.ListJobs(r.Context(), model.JobFilter{ServiceID: id, Limit: serviceJobsLimit})
	return data, err
}

func (s *Server) servicePage(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	data, err := s.serviceDetail(r, id)
	if err != nil {
		s.failLookup(w, err)
		return
	}
	s.render(w, http.StatusOK, "service.html", "layout", data)
}

func (s *Server) serviceJobsFragment(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	jobs, err := s.store.ListJobs(r.Context(), model.JobFilter{ServiceID: id, Limit: serviceJobsLimit})
	if err != nil {
		s.fail(w, err)
		return
	}
	data := s.base(r, "Service")
	data.Jobs = jobs
	s.render(w, http.StatusOK, "service.html", "history", data)
}

func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
	id, ok := pathID(w, r)
	if !ok {
		return
	}
	sess, _ := webauth.SessionFrom(r.Context())
	job, err := s.jobs.Trigger(r.Context(), id, sess.Subject)
	var nu *model.NotUpdatableError
	switch {
	case err == nil || errors.Is(err, model.ErrJobActive):
		http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther)
	case errors.As(err, &nu):
		data, derr := s.serviceDetail(r, id)
		if derr != nil {
			s.failLookup(w, derr)
			return
		}
		data.Error = "Cannot update: " + nu.Reason + "."
		s.render(w, http.StatusConflict, "service.html", "layout", data)
	default:
		s.failLookup(w, err)
	}
}

func (s *Server) jobsData(r *http.Request) (pageData, error) {
	data := s.base(r, "Jobs")
	var err error
	data.Jobs, err = s.store.ListJobs(r.Context(), model.JobFilter{Limit: 100})
	return data, err
}

func (s *Server) jobsPage(w http.ResponseWriter, r *http.Request) {
	data, err := s.jobsData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	s.render(w, http.StatusOK, "jobs.html", "layout", data)
}

func (s *Server) jobsFragment(w http.ResponseWriter, r *http.Request) {
	data, err := s.jobsData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	s.render(w, http.StatusOK, "jobs.html", "jobs", data)
}

// jobData always reads the job from the store, so a fragment fetched after
// a dropped signal still shows the full current output.
func (s *Server) jobData(w http.ResponseWriter, r *http.Request) (pageData, bool) {
	id, ok := pathID(w, r)
	if !ok {
		return pageData{}, false
	}
	job, err := s.store.GetJob(r.Context(), id)
	if err != nil {
		s.failLookup(w, err)
		return pageData{}, false
	}
	data := s.base(r, fmt.Sprintf("Job #%d", job.ID))
	data.Job = job
	return data, true
}

func (s *Server) jobPage(w http.ResponseWriter, r *http.Request) {
	if data, ok := s.jobData(w, r); ok {
		s.render(w, http.StatusOK, "job.html", "layout", data)
	}
}

func (s *Server) jobFragment(w http.ResponseWriter, r *http.Request) {
	if data, ok := s.jobData(w, r); ok {
		s.render(w, http.StatusOK, "job.html", "job", data)
	}
}

func (s *Server) agentData(r *http.Request) (pageData, error) {
	data := s.base(r, "Agents")
	agents, err := s.store.ListAgents(r.Context())
	if err != nil {
		return data, err
	}
	for _, a := range agents {
		data.Agents = append(data.Agents, agentRow{Agent: a, Online: s.hub.Online(a.ID)})
	}
	return data, nil
}

func (s *Server) agents(w http.ResponseWriter, r *http.Request) {
	data, err := s.agentData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	s.render(w, http.StatusOK, "agents.html", "layout", data)
}

func (s *Server) agentsFragment(w http.ResponseWriter, r *http.Request) {
	data, err := s.agentData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	s.render(w, http.StatusOK, "agents.html", "agents", data)
}

func (s *Server) createAgent(w http.ResponseWriter, r *http.Request) {
	name := strings.TrimSpace(r.FormValue("name"))
	if !agentNamePattern.MatchString(name) {
		s.agentsWithError(w, r, http.StatusBadRequest, "Agent names use lowercase letters, digits and dashes.")
		return
	}
	token, err := agenttoken.Generate()
	if err != nil {
		s.fail(w, err)
		return
	}
	agent, err := s.store.CreateAgent(r.Context(), name, agenttoken.Hash(token))
	if errors.Is(err, model.ErrConflict) {
		s.agentsWithError(w, r, http.StatusConflict, fmt.Sprintf("An agent named %q already exists.", name))
		return
	}
	if err != nil {
		s.fail(w, err)
		return
	}
	data := s.base(r, "Agent created")
	data.Created = agent
	data.Snippet = agentSnippet(s.cfg, token)
	w.Header().Set("Cache-Control", "no-store") // the page shows the one-time token
	s.render(w, http.StatusOK, "agent_created.html", "layout", data)
}

func (s *Server) agentsWithError(w http.ResponseWriter, r *http.Request, status int, msg string) {
	data, err := s.agentData(r)
	if err != nil {
		s.fail(w, err)
		return
	}
	data.Error = msg
	s.render(w, status, "agents.html", "layout", data)
}

func (s *Server) revokeAgent(w http.ResponseWriter, r *http.Request) {
	id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
	if err != nil {
		http.Error(w, "invalid agent id", http.StatusBadRequest)
		return
	}
	if err := s.store.RevokeAgent(r.Context(), id, s.now()); err != nil {
		s.fail(w, err)
		return
	}
	s.hub.Disconnect(id)
	s.broker.Publish(events.Services) // the agent's services were marked removed
	s.broker.Publish(events.Jobs)     // the agent's active jobs were failed
	http.Redirect(w, r, "/agents", http.StatusSeeOther)
}

// events streams change signals as Server-Sent Events. htmx's SSE extension
// turns "event: services" into a refetch of the services fragment. The event
// carries no state, so a dropped signal is recovered by the next one.
func (s *Server) events(w http.ResponseWriter, r *http.Request) {
	rc := http.NewResponseController(w)
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no") // stop nginx from buffering the stream

	ch, cancel := s.broker.Subscribe()
	defer cancel()

	if _, err := fmt.Fprint(w, ": connected\n\n"); err != nil {
		return
	}
	if err := rc.Flush(); err != nil {
		s.log.Error("sse_flush_unsupported", "error", err)
		return
	}
	keepalive := time.NewTicker(25 * time.Second)
	defer keepalive.Stop()
	for {
		var msg string
		select {
		case <-r.Context().Done():
			return
		case t := <-ch:
			msg = fmt.Sprintf("event: %s\ndata: changed\n\n", t)
		case <-keepalive.C:
			msg = ": keepalive\n\n"
		}
		if _, err := fmt.Fprint(w, msg); err != nil {
			return
		}
		if err := rc.Flush(); err != nil {
			return
		}
	}
}

func (s *Server) fail(w http.ResponseWriter, err error) {
	s.log.Error("request_failed", "error", err)
	http.Error(w, "internal error", http.StatusInternalServerError)
}

// agentSnippet is the compose file shown once when an agent is added.
func agentSnippet(cfg Config, token string) string {
	return fmt.Sprintf(`services:
  updater-agent:
    image: %s
    entrypoint: ["/usr/local/bin/updater-agent"]
    restart: unless-stopped
    environment:
      MASTER_URL: %s
      MASTER_KEY: %s
      AGENT_TOKEN: %s
      UPDATE_TIMEOUT: 5m
      # Optional registry credentials (a config.json written by docker login):
      # DOCKER_CONFIG: /docker-config
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      # - /path/to/stacks:/path/to/stacks  # mount each stacks directory at the same path, read-write
      # - /root/.docker/config.json:/docker-config/config.json:ro
`, cfg.AgentImage, cfg.AgentURL, cfg.MasterKey, token)
}