Dockerfile

@@ -8,7 +8,7 @@ COPY . .
 ARG VERSION=dev
 ARG COMMIT=unknown
 ARG BUILD_TIME=unknown
-RUN for cmd in server:package-updater agent:updater-agent; do \
+RUN for cmd in server:package-updater master:updater-master agent:updater-agent; do \
       CGO_ENABLED=0 go build \
         -ldflags="-s -w \
           -X main.version=${VERSION} \

cmd/master/main.go

@@ -0,0 +1,137 @@
+// Command updater-master serves the package-updater UI and accepts agent
+// connections.
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/shcizo/package-updater/internal/logging"
+	"github.com/shcizo/package-updater/internal/master/config"
+	"github.com/shcizo/package-updater/internal/master/events"
+	"github.com/shcizo/package-updater/internal/master/hub"
+	"github.com/shcizo/package-updater/internal/master/store"
+	"github.com/shcizo/package-updater/internal/master/web"
+	"github.com/shcizo/package-updater/internal/master/webauth"
+	"github.com/shcizo/package-updater/internal/pin"
+)
+
+// Build info is injected at link time via -ldflags="-X main.version=...".
+var version = "dev"
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintln(os.Stderr, "fatal:", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	cfg, err := config.Load()
+	if err != nil {
+		return err
+	}
+	logger := logging.New(cfg.LogLevel)
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	st, err := store.Open(ctx, cfg.DatabaseURL)
+	if err != nil {
+		return err
+	}
+	defer st.Close()
+
+	key, err := st.LoadOrCreateIdentity(ctx, pin.GenerateKey)
+	if err != nil {
+		return err
+	}
+	cert, err := pin.Certificate(key)
+	if err != nil {
+		return err
+	}
+	fingerprint, err := pin.Fingerprint(key.Public())
+	if err != nil {
+		return err
+	}
+	logger.Info("master_starting", "version", version, "master_key", fingerprint)
+
+	broker := events.NewBroker()
+	agentHub := hub.New(st, broker, logger)
+
+	provider, err := webauth.NewOIDCProvider(ctx, webauth.OIDCConfig{
+		Issuer: cfg.OIDCIssuer, ClientID: cfg.OIDCClientID, ClientSecret: cfg.OIDCClientSecret,
+		RedirectURL: cfg.PublicURL + "/auth/callback",
+	})
+	if err != nil {
+		return err
+	}
+	auth := webauth.New(provider, st, cfg.AllowedGroup, strings.HasPrefix(cfg.PublicURL, "https://"), logger)
+
+	ui, err := web.New(st, agentHub, broker, web.Config{
+		AgentURL: cfg.AgentURL, MasterKey: fingerprint, AgentImage: cfg.AgentImage,
+	}, logger)
+	if err != nil {
+		return err
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
+		if err := st.Ping(r.Context()); err != nil {
+			http.Error(w, "database unreachable", http.StatusServiceUnavailable)
+			return
+		}
+		_, _ = w.Write([]byte("ok"))
+	})
+	mux.HandleFunc("GET /auth/login", auth.Login)
+	mux.HandleFunc("GET /auth/callback", auth.Callback)
+	mux.Handle("POST /auth/logout", auth.Require(http.HandlerFunc(auth.Logout)))
+	ui.Register(mux, auth.Require)
+
+	// BaseContext ties every request context to the signal context, so on
+	// shutdown the long-lived SSE handlers and agent connections return
+	// instead of holding Shutdown until its timeout.
+	baseContext := func(net.Listener) context.Context { return ctx }
+
+	// No WriteTimeout on either server: SSE streams and agent connections
+	// are long-lived by design.
+	webSrv := &http.Server{
+		Addr: ":" + cfg.Port, Handler: mux, ReadHeaderTimeout: 10 * time.Second,
+		BaseContext: baseContext,
+	}
+
+	agentMux := http.NewServeMux()
+	agentMux.Handle("GET /agent", agentHub)
+	agentSrv := &http.Server{
+		Addr: ":" + cfg.AgentPort, Handler: agentMux, ReadHeaderTimeout: 10 * time.Second,
+		TLSConfig:   &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13},
+		BaseContext: baseContext,
+	}
+
+	errCh := make(chan error, 2)
+	go func() { errCh <- webSrv.ListenAndServe() }()
+	go func() { errCh <- agentSrv.ListenAndServeTLS("", "") }()
+	logger.Info("listening", "web", webSrv.Addr, "agents", agentSrv.Addr)
+
+	select {
+	case <-ctx.Done():
+		logger.Info("shutdown_signal")
+	case err := <-errCh:
+		if !errors.Is(err, http.ErrServerClosed) {
+			return err
+		}
+	}
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	_ = webSrv.Shutdown(shutdownCtx)
+	_ = agentSrv.Shutdown(shutdownCtx)
+	return nil
+}

internal/master/config/config.go

@@ -0,0 +1,68 @@
+// Package config loads the master's configuration from environment variables.
+package config
+
+import (
+	"fmt"
+	"os"
+	"strings"
+)
+
+// Config holds the master's runtime configuration.
+type Config struct {
+	DatabaseURL      string
+	PublicURL        string // where people reach the UI, without trailing slash
+	AgentURL         string // what agents dial, shown in the add-agent snippet
+	AgentImage       string
+	OIDCIssuer       string
+	OIDCClientID     string
+	OIDCClientSecret string
+	AllowedGroup     string
+	Port             string
+	AgentPort        string
+	LogLevel         string
+}
+
+// Load reads and validates the master configuration, failing fast on
+// anything missing so a half-configured master never starts.
+func Load() (*Config, error) {
+	cfg := &Config{
+		DatabaseURL:      os.Getenv("DATABASE_URL"),
+		PublicURL:        strings.TrimSuffix(os.Getenv("PUBLIC_URL"), "/"),
+		AgentURL:         os.Getenv("AGENT_URL"),
+		AgentImage:       getenvDefault("AGENT_IMAGE", "gitea.shcizo.se/shcizo/package-updater:latest"),
+		OIDCIssuer:       os.Getenv("OIDC_ISSUER"),
+		OIDCClientID:     os.Getenv("OIDC_CLIENT_ID"),
+		OIDCClientSecret: os.Getenv("OIDC_CLIENT_SECRET"),
+		AllowedGroup:     os.Getenv("ALLOWED_GROUP"),
+		Port:             getenvDefault("PORT", "8080"),
+		AgentPort:        getenvDefault("AGENT_PORT", "8090"),
+		LogLevel:         getenvDefault("LOG_LEVEL", "info"),
+	}
+	for _, r := range []struct{ name, value string }{
+		{"DATABASE_URL", cfg.DatabaseURL},
+		{"PUBLIC_URL", cfg.PublicURL},
+		{"AGENT_URL", cfg.AgentURL},
+		{"OIDC_ISSUER", cfg.OIDCIssuer},
+		{"OIDC_CLIENT_ID", cfg.OIDCClientID},
+		{"OIDC_CLIENT_SECRET", cfg.OIDCClientSecret},
+		{"ALLOWED_GROUP", cfg.AllowedGroup},
+	} {
+		if r.value == "" {
+			return nil, fmt.Errorf("%s is required", r.name)
+		}
+	}
+	if !strings.HasPrefix(cfg.PublicURL, "http://") && !strings.HasPrefix(cfg.PublicURL, "https://") {
+		return nil, fmt.Errorf("PUBLIC_URL must start with http:// or https://")
+	}
+	if !strings.HasPrefix(cfg.AgentURL, "wss://") {
+		return nil, fmt.Errorf("AGENT_URL must start with wss://")
+	}
+	return cfg, nil
+}
+
+func getenvDefault(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}

internal/master/config/config_test.go

@@ -0,0 +1,64 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var required = map[string]string{
+	"DATABASE_URL":       "postgres://u:p@db/pu",
+	"PUBLIC_URL":         "https://updater.example.com",
+	"AGENT_URL":          "wss://10.0.0.5:8090/agent",
+	"OIDC_ISSUER":        "https://auth.example.com/application/o/pu/",
+	"OIDC_CLIENT_ID":     "id",
+	"OIDC_CLIENT_SECRET": "secret",
+	"ALLOWED_GROUP":      "updater-admins",
+}
+
+func setAll(t *testing.T) {
+	for k, v := range required {
+		t.Setenv(k, v)
+	}
+	for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL"} {
+		t.Setenv(k, "")
+	}
+}
+
+func TestLoad_Defaults(t *testing.T) {
+	setAll(t)
+	cfg, err := Load()
+	require.NoError(t, err)
+	assert.Equal(t, "8080", cfg.Port)
+	assert.Equal(t, "8090", cfg.AgentPort)
+	assert.Equal(t, "gitea.shcizo.se/shcizo/package-updater:latest", cfg.AgentImage)
+	assert.Equal(t, "info", cfg.LogLevel)
+	assert.Equal(t, "updater-admins", cfg.AllowedGroup)
+}
+
+func TestLoad_EachRequiredVariable(t *testing.T) {
+	for key := range required {
+		t.Run(key, func(t *testing.T) {
+			setAll(t)
+			t.Setenv(key, "")
+			_, err := Load()
+			assert.ErrorContains(t, err, key+" is required")
+		})
+	}
+}
+
+func TestLoad_URLFormats(t *testing.T) {
+	cases := []struct{ key, value, wantErr string }{
+		{"PUBLIC_URL", "updater.example.com", "PUBLIC_URL must start with http:// or https://"},
+		{"AGENT_URL", "ws://10.0.0.5:8090/agent", "AGENT_URL must start with wss://"},
+	}
+	for _, c := range cases {
+		t.Run(c.key, func(t *testing.T) {
+			setAll(t)
+			t.Setenv(c.key, c.value)
+			_, err := Load()
+			assert.ErrorContains(t, err, c.wantErr)
+		})
+	}
+}