feat(agent): UPDATE_TIMEOUT and DOCKER_CONFIG settings (#7)

cmd/server/main.go

@@ -1,144 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"net/http"
-	"os"
-	"os/signal"
-	"syscall"
-	"time"
-
-	"github.com/docker/docker/client"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/shcizo/package-updater/internal/api"
-	"github.com/shcizo/package-updater/internal/config"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/logging"
-	"github.com/shcizo/package-updater/internal/metrics"
-	"github.com/shcizo/package-updater/internal/updater"
-)
-
-// Build info is injected at link time via -ldflags="-X main.version=..."
-var (
-	version   = "dev"
-	commit    = "unknown"
-	buildTime = "unknown"
-)
-
-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)
-	logger.Info("starting",
-		"version", version, "commit", commit, "port", cfg.Port,
-		"mode", cfg.Mode, "stacks_root", cfg.StacksRoot, "opt_in_label", cfg.OptInLabel,
-	)
-
-	dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
-	if err != nil {
-		return fmt.Errorf("docker client: %w", err)
-	}
-	defer dockerCli.Close()
-
-	var finder api.Finder
-	var exec updater.Executor
-	switch cfg.Mode {
-	case "swarm":
-		finder = discovery.NewSwarm(dockerCli, cfg.OptInLabel)
-		exec = updater.NewSwarmExecutor(dockerCli)
-	case "compose":
-		finder = discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel)
-		exec = updater.NewComposeExecutor()
-	default:
-		// Unreachable: config.Load already validates Mode, but guard
-		// here too rather than silently falling through to a nil
-		// finder/exec pair.
-		return fmt.Errorf("unknown MODE %q", cfg.Mode)
-	}
-
-	reg := prometheus.NewRegistry()
-	m := metrics.New(reg)
-	m.BuildInfo.WithLabelValues(version, commit).Set(1)
-
-	queue := updater.NewQueue(exec, m)
-	queue.Start(context.Background())
-	defer queue.Stop()
-
-	handlers := api.NewHandlers(finder, &submitterAdapter{queue: queue, timeout: cfg.UpdateTimeout}, dockerCli, version, commit, buildTime, m)
-
-	mux := http.NewServeMux()
-	mux.HandleFunc("POST /update", handlers.Update)
-	mux.HandleFunc("GET /healthz", handlers.Healthz)
-	mux.HandleFunc("GET /version", handlers.Version)
-	mux.Handle("GET /metrics", metrics.Handler(reg))
-
-	authed := api.Auth(cfg.APIKey)
-	handler := api.RequestID(api.RequestLogger(logger, m)(routeAuth(mux, authed)))
-
-	srv := &http.Server{
-		Addr:              ":" + cfg.Port,
-		Handler:           handler,
-		ReadHeaderTimeout: 10 * time.Second,
-	}
-
-	errCh := make(chan error, 1)
-	go func() {
-		logger.Info("listening", "addr", srv.Addr)
-		errCh <- srv.ListenAndServe()
-	}()
-
-	sig := make(chan os.Signal, 1)
-	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
-
-	select {
-	case s := <-sig:
-		logger.Info("shutdown_signal", "signal", s.String())
-	case err := <-errCh:
-		if err != nil && err != http.ErrServerClosed {
-			return err
-		}
-	}
-
-	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	_ = srv.Shutdown(shutdownCtx)
-	return nil
-}
-
-// routeAuth applies the bearer-token middleware to /update only.
-// /healthz, /metrics, /version are unauthenticated by design (internal
-// network only; healthcheck and Prometheus scraper need to reach them
-// without secrets).
-func routeAuth(next *http.ServeMux, mw func(http.Handler) http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path == "/update" {
-			mw(next).ServeHTTP(w, r)
-			return
-		}
-		next.ServeHTTP(w, r)
-	})
-}
-
-// submitterAdapter wraps Queue.Submit to inject the per-request timeout
-// from config, satisfying the api.Submitter interface.
-type submitterAdapter struct {
-	queue   *updater.Queue
-	timeout time.Duration
-}
-
-func (s *submitterAdapter) Submit(ctx context.Context, jobs []discovery.Job) []updater.Result {
-	ctx, cancel := context.WithTimeout(ctx, s.timeout)
-	defer cancel()
-	return s.queue.Submit(ctx, jobs)
-}

internal/agent/config/config.go

@@ -2,10 +2,14 @@
 package config
 
 import (
+	"encoding/json"
 	"errors"
+	"fmt"
 	"os"
+	"path/filepath"
 	"regexp"
 	"strings"
+	"time"
 )
 
 var masterKeyPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
@@ -17,8 +21,17 @@ type Config struct {
 	Token     string // AGENT_TOKEN, shown once when the agent was added
 	SelfID    string // SELF_ID, defaults to the hostname (= container ID prefix)
 	LogLevel  string // LOG_LEVEL, default info
+
+	// UpdateTimeout bounds one update job (UPDATE_TIMEOUT, default 5m).
+	UpdateTimeout time.Duration
+	// DockerConfig is the directory holding config.json with registry
+	// credentials (DOCKER_CONFIG). Empty means anonymous registry access.
+	DockerConfig string
 }
 
+// defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset.
+const defaultUpdateTimeout = 5 * time.Minute
+
 // Load reads and validates the agent configuration. It fails fast so a
 // misconfigured agent never starts dialling.
 func Load() (*Config, error) {
@@ -28,6 +41,9 @@ func Load() (*Config, error) {
 		Token:     os.Getenv("AGENT_TOKEN"),
 		SelfID:    os.Getenv("SELF_ID"),
 		LogLevel:  os.Getenv("LOG_LEVEL"),
+
+		UpdateTimeout: defaultUpdateTimeout,
+		DockerConfig:  os.Getenv("DOCKER_CONFIG"),
 	}
 	switch {
 	case cfg.MasterURL == "":
@@ -41,6 +57,18 @@ func Load() (*Config, error) {
 	case cfg.Token == "":
 		return nil, errors.New("AGENT_TOKEN is required")
 	}
+	if v := os.Getenv("UPDATE_TIMEOUT"); v != "" {
+		d, err := time.ParseDuration(v)
+		if err != nil || d <= 0 {
+			return nil, fmt.Errorf("UPDATE_TIMEOUT must be a positive duration such as 5m, got %q", v)
+		}
+		cfg.UpdateTimeout = d
+	}
+	if cfg.DockerConfig != "" {
+		if err := checkDockerConfig(cfg.DockerConfig); err != nil {
+			return nil, err
+		}
+	}
 	if cfg.SelfID == "" {
 		cfg.SelfID, _ = os.Hostname()
 	}
@@ -49,3 +77,17 @@ func Load() (*Config, error) {
 	}
 	return cfg, nil
 }
+
+// checkDockerConfig fails when DOCKER_CONFIG is set but its config.json is
+// missing or not JSON, so a typo does not silently turn into anonymous pulls.
+func checkDockerConfig(dir string) error {
+	path := filepath.Join(dir, "config.json")
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return fmt.Errorf("DOCKER_CONFIG: read %s: %w", path, err)
+	}
+	if !json.Valid(raw) {
+		return fmt.Errorf("DOCKER_CONFIG: %s is not valid JSON", path)
+	}
+	return nil
+}

internal/agent/config/config_test.go

@@ -2,7 +2,9 @@ package config
 
 import (
 	"os"
+	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -16,6 +18,8 @@ func setValid(t *testing.T) {
 	t.Setenv("AGENT_TOKEN", "pua_abc")
 	t.Setenv("SELF_ID", "")
 	t.Setenv("LOG_LEVEL", "")
+	t.Setenv("UPDATE_TIMEOUT", "")
+	t.Setenv("DOCKER_CONFIG", "")
 }
 
 func TestLoad_Valid(t *testing.T) {
@@ -25,7 +29,7 @@ func TestLoad_Valid(t *testing.T) {
 	host, _ := os.Hostname()
 	assert.Equal(t, &Config{
 		MasterURL: "wss://10.0.0.5:8090/agent", MasterKey: validKey, Token: "pua_abc",
-		SelfID: host, LogLevel: "info",
+		SelfID: host, LogLevel: "info", UpdateTimeout: 5 * time.Minute,
 	}, cfg)
 }
 
@@ -48,3 +52,45 @@ func TestLoad_Invalid(t *testing.T) {
 		})
 	}
 }
+
+func TestLoad_JobSettings(t *testing.T) {
+	withConfig := func(t *testing.T, body string) string {
+		dir := t.TempDir()
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(body), 0o600))
+		return dir
+	}
+	cases := []struct {
+		name      string
+		timeout   string
+		dockerCfg func(t *testing.T) string
+		wantTime  time.Duration
+		wantErr   string
+	}{
+		{name: "defaults", wantTime: 5 * time.Minute},
+		{name: "custom timeout", timeout: "90s", wantTime: 90 * time.Second},
+		{name: "zero timeout", timeout: "0", wantErr: "UPDATE_TIMEOUT"},
+		{name: "garbage timeout", timeout: "abc", wantErr: "UPDATE_TIMEOUT"},
+		{name: "docker config without file", dockerCfg: func(t *testing.T) string { return t.TempDir() }, wantErr: "DOCKER_CONFIG"},
+		{name: "docker config not json", dockerCfg: func(t *testing.T) string { return withConfig(t, "nope") }, wantErr: "not valid JSON"},
+		{name: "docker config ok", dockerCfg: func(t *testing.T) string { return withConfig(t, `{"auths":{}}`) }, wantTime: 5 * time.Minute},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			setValid(t)
+			t.Setenv("UPDATE_TIMEOUT", c.timeout)
+			dir := ""
+			if c.dockerCfg != nil {
+				dir = c.dockerCfg(t)
+				t.Setenv("DOCKER_CONFIG", dir)
+			}
+			cfg, err := Load()
+			if c.wantErr != "" {
+				assert.ErrorContains(t, err, c.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, c.wantTime, cfg.UpdateTimeout)
+			assert.Equal(t, dir, cfg.DockerConfig)
+		})
+	}
+}

internal/agent/inventory/collector.go

@@ -18,10 +18,11 @@ import (
 )
 
 const (
-	labelProject    = "com.docker.compose.project"
-	labelService    = "com.docker.compose.service"
-	labelWorkingDir = "com.docker.compose.project.working_dir"
-	labelSwarmTask  = "com.docker.swarm.service.id"
+	labelProject     = "com.docker.compose.project"
+	labelService     = "com.docker.compose.service"
+	labelWorkingDir  = "com.docker.compose.project.working_dir"
+	labelConfigFiles = "com.docker.compose.project.config_files"
+	labelSwarmTask   = "com.docker.swarm.service.id"
 )
 
 // DockerClient is the subset of the Docker SDK the collector needs.
@@ -148,6 +149,7 @@ func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests
 		svc.ComposeProject = project
 		svc.ComposeService = service
 		svc.WorkingDir = ct.Labels[labelWorkingDir]
+		svc.ConfigFiles = ct.Labels[labelConfigFiles]
 		return svc
 	}
 	svc.Kind = protocol.KindContainer

internal/agent/inventory/collector_test.go

@@ -200,3 +200,12 @@ func TestHello(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "cluster-1"}, got)
 }
+
+func TestSnapshot_ComposeConfigFilesIsRawLabel(t *testing.T) {
+	ct := composeContainer("abc123", "slusk", "web", "slusk:1", "sha256:img1", "running")
+	ct.Labels[labelConfigFiles] = "/s/a/compose.yaml,/s/a/compose.override.yaml"
+	got, err := New(&fakeDocker{containers: []container.Summary{ct}}, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "/s/a/compose.yaml,/s/a/compose.override.yaml", got[0].ConfigFiles)
+}

internal/api/handlers.go

@@ -1,148 +0,0 @@
-package api
-
-import (
-	"context"
-	"encoding/json"
-	"net/http"
-
-	"github.com/docker/docker/api/types"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/logging"
-	"github.com/shcizo/package-updater/internal/metrics"
-	"github.com/shcizo/package-updater/internal/updater"
-)
-
-// Finder is the discovery interface used by the update handler.
-type Finder interface {
-	FindJobs(ctx context.Context, image string) ([]discovery.Job, error)
-}
-
-// Submitter is the queue interface used by the update handler.
-type Submitter interface {
-	Submit(ctx context.Context, jobs []discovery.Job) []updater.Result
-}
-
-// Pinger pings the Docker daemon for the healthcheck.
-type Pinger interface {
-	Ping(ctx context.Context) (types.Ping, error)
-}
-
-// Handlers wires the HTTP endpoints to the rest of the service.
-type Handlers struct {
-	finder    Finder
-	submitter Submitter
-	pinger    Pinger
-	version   string
-	commit    string
-	buildTime string
-	metrics   *metrics.Metrics
-}
-
-// NewHandlers constructs a Handlers value with all dependencies injected.
-// m may be nil, in which case metrics recording is silently skipped.
-func NewHandlers(f Finder, s Submitter, p Pinger, version, commit, buildTime string, m *metrics.Metrics) *Handlers {
-	return &Handlers{finder: f, submitter: s, pinger: p, version: version, commit: commit, buildTime: buildTime, metrics: m}
-}
-
-// Update implements POST /update.
-func (h *Handlers) Update(w http.ResponseWriter, r *http.Request) {
-	var req UpdateRequest
-	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
-		writeJSONError(w, http.StatusBadRequest, "invalid JSON body")
-		return
-	}
-	if req.Image == "" {
-		writeJSONError(w, http.StatusBadRequest, "image is required")
-		return
-	}
-
-	requestedImage := req.Image
-	if req.Tag != "" {
-		requestedImage = req.Image + ":" + req.Tag
-	}
-
-	jobs, err := h.finder.FindJobs(r.Context(), requestedImage)
-	if err != nil {
-		writeJSONError(w, http.StatusInternalServerError, "discovery failed: "+err.Error())
-		return
-	}
-
-	resp := UpdateResponse{
-		RequestID: logging.RequestIDFrom(r.Context()),
-		Image:     req.Image,
-		Tag:       req.Tag,
-		Matched:   len(jobs),
-		Results:   []ResultDTO{},
-	}
-
-	if len(jobs) == 0 {
-		writeJSON(w, http.StatusOK, resp)
-		return
-	}
-
-	results := h.submitter.Submit(r.Context(), jobs)
-	updated, failed := 0, 0
-	for _, res := range results {
-		composeFile := ""
-		if len(res.Job.ConfigFiles) > 0 {
-			composeFile = res.Job.ConfigFiles[0]
-		}
-		resp.Results = append(resp.Results, ResultDTO{
-			Project:     res.Job.Project,
-			Service:     res.Job.Service,
-			ComposeFile: composeFile,
-			ServiceID:   res.Job.ServiceID,
-			Status:      string(res.Status),
-			Error:       res.Error,
-			DurationMs:  res.DurationMs,
-		})
-		if res.Status == updater.StatusUpdated {
-			updated++
-		} else {
-			failed++
-		}
-	}
-
-	switch {
-	case failed == 0:
-		writeJSON(w, http.StatusOK, resp)
-	case updated == 0:
-		writeJSON(w, http.StatusInternalServerError, resp)
-	default:
-		writeJSON(w, http.StatusMultiStatus, resp)
-	}
-}
-
-// Healthz implements GET /healthz.
-func (h *Handlers) Healthz(w http.ResponseWriter, r *http.Request) {
-	if _, err := h.pinger.Ping(r.Context()); err != nil {
-		if h.metrics != nil {
-			h.metrics.DockerPingUp.Set(0)
-		}
-		writeJSON(w, http.StatusServiceUnavailable, HealthResponse{Status: "unhealthy", Docker: "unreachable"})
-		return
-	}
-	if h.metrics != nil {
-		h.metrics.DockerPingUp.Set(1)
-	}
-	writeJSON(w, http.StatusOK, HealthResponse{Status: "ok", Docker: "ok"})
-}
-
-// Version implements GET /version.
-func (h *Handlers) Version(w http.ResponseWriter, _ *http.Request) {
-	writeJSON(w, http.StatusOK, VersionResponse{
-		Version:   h.version,
-		Commit:    h.commit,
-		BuildTime: h.buildTime,
-	})
-}
-
-func writeJSON(w http.ResponseWriter, code int, body any) {
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(code)
-	_ = json.NewEncoder(w).Encode(body)
-}
-
-func writeJSONError(w http.ResponseWriter, code int, msg string) {
-	writeJSON(w, code, map[string]string{"error": msg})
-}

internal/api/handlers_test.go

@@ -1,216 +0,0 @@
-package api_test
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-
-	"github.com/docker/docker/api/types"
-	"github.com/shcizo/package-updater/internal/api"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/updater"
-	"github.com/stretchr/testify/require"
-)
-
-type fakeFinder struct {
-	jobs []discovery.Job
-	err  error
-
-	// gotImage records the image argument passed to FindJobs, for
-	// assertions on the exact reference the handler built.
-	gotImage string
-	// swarmShaped, when true, makes FindJobs return a Job whose Image
-	// field carries the received image argument, simulating
-	// discovery.SwarmDiscovery.FindJobs.
-	swarmShaped bool
-}
-
-func (f *fakeFinder) FindJobs(_ context.Context, image string) ([]discovery.Job, error) {
-	f.gotImage = image
-	if f.swarmShaped {
-		return []discovery.Job{{Service: "myapp_web", ServiceID: "svc1", Image: image}}, f.err
-	}
-	return f.jobs, f.err
-}
-
-type fakeSubmitter struct {
-	results []updater.Result
-}
-
-func (f *fakeSubmitter) Submit(_ context.Context, jobs []discovery.Job) []updater.Result {
-	if f.results != nil {
-		return f.results
-	}
-	out := make([]updater.Result, len(jobs))
-	for i, j := range jobs {
-		out[i] = updater.Result{Job: j, Status: updater.StatusUpdated}
-	}
-	return out
-}
-
-type fakePinger struct{ err error }
-
-func (f *fakePinger) Ping(_ context.Context) (types.Ping, error) {
-	return types.Ping{}, f.err
-}
-
-func decode[T any](t *testing.T, body io.Reader) T {
-	t.Helper()
-	var v T
-	require.NoError(t, json.NewDecoder(body).Decode(&v))
-	return v
-}
-
-func TestUpdate_ValidationError(t *testing.T) {
-	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	req := httptest.NewRequest(http.MethodPost, "/update",
-		strings.NewReader(`{"tag":"v1.2.3"}`))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusBadRequest, w.Code)
-}
-
-func TestUpdate_BadJSON(t *testing.T) {
-	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`not json`))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusBadRequest, w.Code)
-}
-
-func TestUpdate_DiscoveryFailureReturns500(t *testing.T) {
-	finder := &fakeFinder{err: errors.New("daemon unreachable")}
-	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusInternalServerError, w.Code)
-}
-
-func TestUpdate_ZeroMatchesReturns200(t *testing.T) {
-	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusOK, w.Code)
-	resp := decode[api.UpdateResponse](t, w.Body)
-	require.Equal(t, 0, resp.Matched)
-}
-
-func TestUpdate_AllSucceeded200(t *testing.T) {
-	finder := &fakeFinder{jobs: []discovery.Job{
-		{Project: "p", Service: "s", WorkingDir: "/x", ConfigFiles: []string{"/x/c.yml"}},
-	}}
-	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "r/x", Tag: "v1"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusOK, w.Code)
-	resp := decode[api.UpdateResponse](t, w.Body)
-	require.Equal(t, 1, resp.Matched)
-	require.Equal(t, "updated", resp.Results[0].Status)
-	require.Equal(t, "/x/c.yml", resp.Results[0].ComposeFile)
-}
-
-func TestUpdate_MixedReturns207(t *testing.T) {
-	jobs := []discovery.Job{
-		{Project: "p1", Service: "s", ConfigFiles: []string{"/x/c.yml"}},
-		{Project: "p2", Service: "s", ConfigFiles: []string{"/y/c.yml"}},
-	}
-	finder := &fakeFinder{jobs: jobs}
-	submitter := &fakeSubmitter{results: []updater.Result{
-		{Job: jobs[0], Status: updater.StatusUpdated},
-		{Job: jobs[1], Status: updater.StatusFailed, Error: "boom"},
-	}}
-	h := api.NewHandlers(finder, submitter, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusMultiStatus, w.Code)
-}
-
-func TestUpdate_AllFailedReturns500(t *testing.T) {
-	jobs := []discovery.Job{{Project: "p", Service: "s", ConfigFiles: []string{"/x/c.yml"}}}
-	finder := &fakeFinder{jobs: jobs}
-	submitter := &fakeSubmitter{results: []updater.Result{
-		{Job: jobs[0], Status: updater.StatusFailed, Error: "boom"},
-	}}
-	h := api.NewHandlers(finder, submitter, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusInternalServerError, w.Code)
-}
-
-func TestHealthz_OKWhenDockerUp(t *testing.T) {
-	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
-	w := httptest.NewRecorder()
-	h.Healthz(w, req)
-	require.Equal(t, http.StatusOK, w.Code)
-}
-
-func TestHealthz_503WhenDockerDown(t *testing.T) {
-	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{err: errors.New("ping fail")}, "v0.0.0", "abc", "now", nil)
-	req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
-	w := httptest.NewRecorder()
-	h.Healthz(w, req)
-	require.Equal(t, http.StatusServiceUnavailable, w.Code)
-}
-
-func TestUpdate_FindJobsCalledWithTaggedImage(t *testing.T) {
-	finder := &fakeFinder{}
-	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp", Tag: "v2"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, "registry.example.com/myapp:v2", finder.gotImage)
-}
-
-func TestUpdate_FindJobsCalledWithBareImageWhenNoTag(t *testing.T) {
-	finder := &fakeFinder{}
-	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, "registry.example.com/myapp", finder.gotImage)
-}
-
-func TestUpdate_SwarmModeThreadsTagThroughDiscovery(t *testing.T) {
-	finder := &fakeFinder{swarmShaped: true}
-	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
-	body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp", Tag: "v2"})
-	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
-	w := httptest.NewRecorder()
-	h.Update(w, req)
-	require.Equal(t, http.StatusOK, w.Code)
-	// The fake Finder's returned Job.Image (as SwarmDiscovery would build
-	// it) reflects the image argument it received from the handler.
-	// Asserting on the captured argument proves the full request ->
-	// discovery flow carries the tag through to what would reach the
-	// SwarmExecutor.
-	require.Equal(t, "registry.example.com/myapp:v2", finder.gotImage)
-}
-
-func TestVersion(t *testing.T) {
-	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v1.2.3", "abcdef", "2026-05-22T00:00:00Z", nil)
-	req := httptest.NewRequest(http.MethodGet, "/version", nil)
-	w := httptest.NewRecorder()
-	h.Version(w, req)
-	require.Equal(t, http.StatusOK, w.Code)
-	resp := decode[api.VersionResponse](t, w.Body)
-	require.Equal(t, "v1.2.3", resp.Version)
-}

internal/api/middleware.go

@@ -1,112 +0,0 @@
-// Package api contains HTTP handlers, middleware, and request/response DTOs.
-package api
-
-import (
-	"crypto/rand"
-	"crypto/subtle"
-	"fmt"
-	"log/slog"
-	"net/http"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/shcizo/package-updater/internal/logging"
-	"github.com/shcizo/package-updater/internal/metrics"
-)
-
-// Auth returns middleware that requires a matching bearer token.
-// Compares with constant-time to defeat timing attacks.
-func Auth(token string) func(http.Handler) http.Handler {
-	tokenBytes := []byte(token)
-	return func(next http.Handler) http.Handler {
-		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			h := r.Header.Get("Authorization")
-			const prefix = "Bearer "
-			if !strings.HasPrefix(h, prefix) {
-				writeAuthError(w)
-				return
-			}
-			provided := []byte(strings.TrimPrefix(h, prefix))
-			if len(provided) == 0 ||
-				subtle.ConstantTimeCompare(provided, tokenBytes) != 1 {
-				writeAuthError(w)
-				return
-			}
-			next.ServeHTTP(w, r)
-		})
-	}
-}
-
-func writeAuthError(w http.ResponseWriter) {
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(http.StatusUnauthorized)
-	_, _ = w.Write([]byte(`{"error":"unauthorized"}`))
-}
-
-// newUUID generates a random UUID v4 string using crypto/rand.
-func newUUID() string {
-	var b [16]byte
-	_, _ = rand.Read(b[:])
-	b[6] = (b[6] & 0x0f) | 0x40 // version 4
-	b[8] = (b[8] & 0x3f) | 0x80 // variant bits
-	return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x",
-		b[0:4], b[4:6], b[6:8], b[8:10], b[10:16])
-}
-
-// RequestID middleware ensures every request has an X-Request-ID
-// header (generated if absent) and stores it in the request context.
-func RequestID(next http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		id := r.Header.Get("X-Request-ID")
-		if id == "" {
-			id = newUUID()
-		}
-		w.Header().Set("X-Request-ID", id)
-		ctx := logging.WithRequestID(r.Context(), id)
-		next.ServeHTTP(w, r.WithContext(ctx))
-	})
-}
-
-// RequestLogger emits a structured access-log line per request. m may be nil,
-// in which case metrics recording is silently skipped.
-func RequestLogger(base *slog.Logger, m *metrics.Metrics) func(http.Handler) http.Handler {
-	return func(next http.Handler) http.Handler {
-		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			start := time.Now()
-			sw := &statusWriter{ResponseWriter: w, status: http.StatusOK}
-			next.ServeHTTP(sw, r)
-			logger := logging.FromContext(r.Context(), base)
-			logger.Info("http_request",
-				"method", r.Method,
-				"path", r.URL.Path,
-				"status", sw.status,
-				"duration_ms", time.Since(start).Milliseconds(),
-				"client_ip", clientIP(r),
-			)
-			if m != nil {
-				m.HTTPRequests.WithLabelValues(r.URL.Path, strconv.Itoa(sw.status)).Inc()
-			}
-		})
-	}
-}
-
-type statusWriter struct {
-	http.ResponseWriter
-	status int
-}
-
-func (s *statusWriter) WriteHeader(code int) {
-	s.status = code
-	s.ResponseWriter.WriteHeader(code)
-}
-
-func clientIP(r *http.Request) string {
-	if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
-		if comma := strings.Index(xff, ","); comma >= 0 {
-			return strings.TrimSpace(xff[:comma])
-		}
-		return xff
-	}
-	return r.RemoteAddr
-}

internal/api/middleware_test.go

@@ -1,92 +0,0 @@
-package api_test
-
-import (
-	"io"
-	"log/slog"
-	"net/http"
-	"net/http/httptest"
-	"testing"
-
-	"github.com/shcizo/package-updater/internal/api"
-	"github.com/shcizo/package-updater/internal/logging"
-	"github.com/stretchr/testify/require"
-)
-
-func newTestLogger() *slog.Logger {
-	return slog.New(slog.NewJSONHandler(io.Discard, nil))
-}
-
-func TestAuth_AllowsMatchingToken(t *testing.T) {
-	called := false
-	h := api.Auth("secret")(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		called = true
-		w.WriteHeader(http.StatusOK)
-	}))
-	req := httptest.NewRequest(http.MethodPost, "/update", nil)
-	req.Header.Set("Authorization", "Bearer secret")
-	w := httptest.NewRecorder()
-	h.ServeHTTP(w, req)
-	require.True(t, called)
-	require.Equal(t, http.StatusOK, w.Code)
-}
-
-func TestAuth_Rejects(t *testing.T) {
-	cases := []struct{ name, header string }{
-		{"missing", ""},
-		{"wrong scheme", "Token secret"},
-		{"wrong value", "Bearer nope"},
-		{"empty bearer", "Bearer "},
-	}
-	for _, c := range cases {
-		t.Run(c.name, func(t *testing.T) {
-			h := api.Auth("secret")(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-				t.Fatal("handler must not be called")
-			}))
-			req := httptest.NewRequest(http.MethodPost, "/update", nil)
-			if c.header != "" {
-				req.Header.Set("Authorization", c.header)
-			}
-			w := httptest.NewRecorder()
-			h.ServeHTTP(w, req)
-			require.Equal(t, http.StatusUnauthorized, w.Code)
-		})
-	}
-}
-
-func TestRequestID_GeneratesIfMissing(t *testing.T) {
-	var seenID string
-	h := api.RequestID(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
-		seenID = logging.RequestIDFrom(r.Context())
-	}))
-	req := httptest.NewRequest(http.MethodPost, "/update", nil)
-	w := httptest.NewRecorder()
-	h.ServeHTTP(w, req)
-	require.NotEmpty(t, seenID)
-	require.Equal(t, seenID, w.Header().Get("X-Request-ID"))
-}
-
-func TestRequestID_UsesIncoming(t *testing.T) {
-	var seenID string
-	h := api.RequestID(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
-		seenID = logging.RequestIDFrom(r.Context())
-	}))
-	req := httptest.NewRequest(http.MethodPost, "/update", nil)
-	req.Header.Set("X-Request-ID", "given-id")
-	w := httptest.NewRecorder()
-	h.ServeHTTP(w, req)
-	require.Equal(t, "given-id", seenID)
-}
-
-func TestRequestLogger_LogsAndDelegates(t *testing.T) {
-	logger := newTestLogger()
-	called := false
-	h := api.RequestLogger(logger, nil)(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		called = true
-		w.WriteHeader(http.StatusTeapot)
-	}))
-	req := httptest.NewRequest(http.MethodPost, "/update", nil)
-	w := httptest.NewRecorder()
-	h.ServeHTTP(w, req)
-	require.True(t, called)
-	require.Equal(t, http.StatusTeapot, w.Code)
-}

internal/api/swarm_result_test.go

@@ -1,48 +0,0 @@
-package api_test
-
-import (
-	"context"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-
-	"github.com/docker/docker/api/types"
-	"github.com/shcizo/package-updater/internal/api"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/updater"
-	"github.com/stretchr/testify/require"
-)
-
-type swarmResultFinder struct{ jobs []discovery.Job }
-
-func (f *swarmResultFinder) FindJobs(_ context.Context, _ string) ([]discovery.Job, error) {
-	return f.jobs, nil
-}
-
-type swarmResultSubmitter struct{ results []updater.Result }
-
-func (s *swarmResultSubmitter) Submit(_ context.Context, _ []discovery.Job) []updater.Result {
-	return s.results
-}
-
-type swarmResultPinger struct{}
-
-func (swarmResultPinger) Ping(_ context.Context) (types.Ping, error) { return types.Ping{}, nil }
-
-func TestUpdate_SwarmJobPopulatesServiceID(t *testing.T) {
-	job := discovery.Job{Service: "myapp_web", ServiceID: "svc123", Image: "registry.example.com/myapp:v2"}
-	finder := &swarmResultFinder{jobs: []discovery.Job{job}}
-	submitter := &swarmResultSubmitter{results: []updater.Result{
-		{Job: job, Status: updater.StatusUpdated, DurationMs: 42},
-	}}
-	h := api.NewHandlers(finder, submitter, swarmResultPinger{}, "v", "c", "b", nil)
-
-	req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"registry.example.com/myapp"}`))
-	rec := httptest.NewRecorder()
-	h.Update(rec, req)
-
-	require.Equal(t, http.StatusOK, rec.Code)
-	require.Contains(t, rec.Body.String(), `"service_id":"svc123"`)
-	require.Contains(t, rec.Body.String(), `"service":"myapp_web"`)
-}

internal/api/types.go

@@ -1,40 +0,0 @@
-package api
-
-// UpdateRequest is the body of POST /update.
-type UpdateRequest struct {
-	Image string `json:"image"`
-	Tag   string `json:"tag,omitempty"`
-}
-
-// UpdateResponse is the body of POST /update.
-type UpdateResponse struct {
-	RequestID string      `json:"request_id"`
-	Image     string      `json:"image"`
-	Tag       string      `json:"tag,omitempty"`
-	Matched   int         `json:"matched"`
-	Results   []ResultDTO `json:"results"`
-}
-
-// ResultDTO is one row in UpdateResponse.Results.
-type ResultDTO struct {
-	Project     string `json:"project"`
-	Service     string `json:"service"`
-	ComposeFile string `json:"compose_file"`
-	ServiceID   string `json:"service_id,omitempty"`
-	Status      string `json:"status"`
-	Error       string `json:"error,omitempty"`
-	DurationMs  int64  `json:"duration_ms"`
-}
-
-// HealthResponse is the body of GET /healthz.
-type HealthResponse struct {
-	Status string `json:"status"`
-	Docker string `json:"docker"`
-}
-
-// VersionResponse is the body of GET /version.
-type VersionResponse struct {
-	Version   string `json:"version"`
-	Commit    string `json:"commit"`
-	BuildTime string `json:"build_time"`
-}

internal/config/config.go

@@ -1,58 +0,0 @@
-// Package config loads service configuration from environment variables.
-package config
-
-import (
-	"errors"
-	"fmt"
-	"os"
-	"time"
-)
-
-// Config holds all runtime configuration for the service.
-type Config struct {
-	APIKey        string
-	StacksRoot    string
-	Port          string
-	LogLevel      string
-	UpdateTimeout time.Duration
-	OptInLabel    string
-	Mode          string
-}
-
-// Load reads configuration from environment variables, applies defaults,
-// and validates required fields. Returns an error if validation fails so
-// the service can fail-fast at startup.
-func Load() (*Config, error) {
-	cfg := &Config{
-		APIKey:     os.Getenv("UPDATER_API_KEY"),
-		StacksRoot: getenvDefault("STACKS_ROOT", "/home/shcizo/self-hosted"),
-		Port:       getenvDefault("PORT", "8080"),
-		LogLevel:   getenvDefault("LOG_LEVEL", "info"),
-		OptInLabel: getenvDefault("OPT_IN_LABEL", "se.shcizo.auto-update"),
-		Mode:       getenvDefault("MODE", "compose"),
-	}
-
-	if cfg.APIKey == "" {
-		return nil, errors.New("UPDATER_API_KEY is required")
-	}
-
-	if cfg.Mode != "compose" && cfg.Mode != "swarm" {
-		return nil, fmt.Errorf("MODE %q is invalid: must be %q or %q", cfg.Mode, "compose", "swarm")
-	}
-
-	timeoutStr := getenvDefault("UPDATE_TIMEOUT", "5m")
-	d, err := time.ParseDuration(timeoutStr)
-	if err != nil {
-		return nil, fmt.Errorf("UPDATE_TIMEOUT %q is not a valid duration: %w", timeoutStr, err)
-	}
-	cfg.UpdateTimeout = d
-
-	return cfg, nil
-}
-
-func getenvDefault(key, fallback string) string {
-	if v := os.Getenv(key); v != "" {
-		return v
-	}
-	return fallback
-}

internal/config/config_test.go

@@ -1,82 +0,0 @@
-package config_test
-
-import (
-	"testing"
-	"time"
-
-	"github.com/shcizo/package-updater/internal/config"
-	"github.com/stretchr/testify/require"
-)
-
-func TestLoad_RequiresAPIKey(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "")
-	_, err := config.Load()
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "UPDATER_API_KEY")
-}
-
-func TestLoad_AppliesDefaults(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "secret")
-	t.Setenv("STACKS_ROOT", "")
-	t.Setenv("PORT", "")
-	t.Setenv("LOG_LEVEL", "")
-	t.Setenv("UPDATE_TIMEOUT", "")
-	t.Setenv("OPT_IN_LABEL", "")
-
-	cfg, err := config.Load()
-	require.NoError(t, err)
-	require.Equal(t, "secret", cfg.APIKey)
-	require.Equal(t, "/home/shcizo/self-hosted", cfg.StacksRoot)
-	require.Equal(t, "8080", cfg.Port)
-	require.Equal(t, "info", cfg.LogLevel)
-	require.Equal(t, 5*time.Minute, cfg.UpdateTimeout)
-	require.Equal(t, "se.shcizo.auto-update", cfg.OptInLabel)
-}
-
-func TestLoad_OverridesViaEnv(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "secret")
-	t.Setenv("STACKS_ROOT", "/srv/stacks")
-	t.Setenv("PORT", "9090")
-	t.Setenv("LOG_LEVEL", "debug")
-	t.Setenv("UPDATE_TIMEOUT", "30s")
-	t.Setenv("OPT_IN_LABEL", "io.example.update")
-
-	cfg, err := config.Load()
-	require.NoError(t, err)
-	require.Equal(t, "/srv/stacks", cfg.StacksRoot)
-	require.Equal(t, "9090", cfg.Port)
-	require.Equal(t, "debug", cfg.LogLevel)
-	require.Equal(t, 30*time.Second, cfg.UpdateTimeout)
-	require.Equal(t, "io.example.update", cfg.OptInLabel)
-}
-
-func TestLoad_InvalidTimeoutErrors(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "secret")
-	t.Setenv("UPDATE_TIMEOUT", "not-a-duration")
-	_, err := config.Load()
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "UPDATE_TIMEOUT")
-}
-
-func TestLoad_DefaultsToComposeMode(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "secret")
-	cfg, err := config.Load()
-	require.NoError(t, err)
-	require.Equal(t, "compose", cfg.Mode)
-}
-
-func TestLoad_AcceptsSwarmMode(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "secret")
-	t.Setenv("MODE", "swarm")
-	cfg, err := config.Load()
-	require.NoError(t, err)
-	require.Equal(t, "swarm", cfg.Mode)
-}
-
-func TestLoad_RejectsInvalidMode(t *testing.T) {
-	t.Setenv("UPDATER_API_KEY", "secret")
-	t.Setenv("MODE", "kubernetes")
-	_, err := config.Load()
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "MODE")
-}

internal/discovery/discovery.go

@@ -1,96 +0,0 @@
-package discovery
-
-import (
-	"context"
-	"fmt"
-	"sort"
-	"strings"
-
-	"github.com/docker/docker/api/types/container"
-)
-
-// Job describes a single (project, service, config_files) update to execute.
-type Job struct {
-	Project     string
-	Service     string
-	WorkingDir  string
-	ConfigFiles []string
-	// Image is the full image reference the update request asked for.
-	// Compose jobs ignore it (compose.yml already pins the reference to
-	// pull); Swarm jobs need it to know what to set on the service spec.
-	Image string
-	// ServiceID is the Swarm service ID. Empty for Compose jobs.
-	ServiceID string
-	// Refused is true when the WorkingDir falls outside STACKS_ROOT.
-	// The job is returned so the caller can surface a per-job "refused"
-	// result, but it MUST NOT be executed.
-	Refused       bool
-	RefusedReason string
-}
-
-// Discovery orchestrates "given an image, what jobs should we enqueue?".
-type Discovery struct {
-	cli        DockerClient
-	stacksRoot string
-	optInLabel string
-}
-
-// New returns a Discovery bound to the given Docker client and config.
-func New(cli DockerClient, stacksRoot, optInLabel string) *Discovery {
-	return &Discovery{cli: cli, stacksRoot: stacksRoot, optInLabel: optInLabel}
-}
-
-// FindJobs lists running containers, filters by image match + opt-in label,
-// extracts Compose info, applies the path safety check, and deduplicates.
-func (d *Discovery) FindJobs(ctx context.Context, image string) ([]Job, error) {
-	all, err := d.cli.ContainerList(ctx, container.ListOptions{All: true})
-	if err != nil {
-		return nil, fmt.Errorf("docker container list: %w", err)
-	}
-
-	seen := make(map[string]struct{})
-	var jobs []Job
-
-	for _, c := range all {
-		if !ImagesMatch(image, c.Image) {
-			continue
-		}
-		if !HasOptIn(c.Labels, d.optInLabel) {
-			continue
-		}
-		cl, err := ParseComposeLabels(c.Labels)
-		if err != nil {
-			continue
-		}
-
-		refused := !IsInsideRoot(d.stacksRoot, cl.WorkingDir)
-		reason := ""
-		if refused {
-			reason = fmt.Sprintf("working_dir %q outside STACKS_ROOT %q", cl.WorkingDir, d.stacksRoot)
-		}
-
-		key := dedupKey(cl)
-		if _, dup := seen[key]; dup {
-			continue
-		}
-		seen[key] = struct{}{}
-
-		jobs = append(jobs, Job{
-			Project:       cl.Project,
-			Service:       cl.Service,
-			WorkingDir:    cl.WorkingDir,
-			ConfigFiles:   cl.ConfigFiles,
-			Image:         image,
-			Refused:       refused,
-			RefusedReason: reason,
-		})
-	}
-
-	return jobs, nil
-}
-
-func dedupKey(cl ComposeLabels) string {
-	files := append([]string(nil), cl.ConfigFiles...)
-	sort.Strings(files)
-	return cl.Project + "|" + cl.Service + "|" + strings.Join(files, ",")
-}

internal/discovery/discovery_test.go

@@ -1,172 +0,0 @@
-package discovery_test
-
-import (
-	"context"
-	"errors"
-	"testing"
-
-	"github.com/docker/docker/api/types"
-	"github.com/docker/docker/api/types/container"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/stretchr/testify/require"
-)
-
-type fakeDockerClient struct {
-	containers []types.Container
-	err        error
-}
-
-func (f *fakeDockerClient) ContainerList(_ context.Context, _ container.ListOptions) ([]types.Container, error) {
-	return f.containers, f.err
-}
-
-func (f *fakeDockerClient) Ping(_ context.Context) (types.Ping, error) {
-	return types.Ping{}, nil
-}
-
-func mkContainer(image string, labels map[string]string) types.Container {
-	return types.Container{Image: image, Labels: labels}
-}
-
-func mkComposeLabels(project, service, workingDir, configFile string, optIn bool) map[string]string {
-	m := map[string]string{
-		"com.docker.compose.project":              project,
-		"com.docker.compose.service":              service,
-		"com.docker.compose.project.working_dir":  workingDir,
-		"com.docker.compose.project.config_files": configFile,
-	}
-	if optIn {
-		m["se.shcizo.auto-update"] = "true"
-	}
-	return m
-}
-
-func TestFindJobs_MatchAndOptIn(t *testing.T) {
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
-			"myapp-prod", "web",
-			"/home/shcizo/self-hosted/myapp-prod",
-			"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
-			true,
-		)),
-		mkContainer("registry.example.com/other:v1", mkComposeLabels(
-			"other", "web",
-			"/home/shcizo/self-hosted/other",
-			"/home/shcizo/self-hosted/other/docker-compose.yml",
-			true,
-		)),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Len(t, jobs, 1)
-	require.Equal(t, "myapp-prod", jobs[0].Project)
-	require.Equal(t, "web", jobs[0].Service)
-	require.Equal(t, "/home/shcizo/self-hosted/myapp-prod", jobs[0].WorkingDir)
-}
-
-func TestFindJobs_SkipsWithoutOptIn(t *testing.T) {
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
-			"myapp-prod", "web",
-			"/home/shcizo/self-hosted/myapp-prod",
-			"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
-			false,
-		)),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Empty(t, jobs)
-}
-
-func TestFindJobs_DedupReplicas(t *testing.T) {
-	labels := mkComposeLabels(
-		"myapp", "web",
-		"/home/shcizo/self-hosted/myapp",
-		"/home/shcizo/self-hosted/myapp/docker-compose.yml",
-		true,
-	)
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", labels),
-		mkContainer("registry.example.com/myapp:v1", labels),
-		mkContainer("registry.example.com/myapp:v1", labels),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Len(t, jobs, 1)
-}
-
-func TestFindJobs_OutsideRootProducesRefusedJob(t *testing.T) {
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
-			"myapp", "web",
-			"/opt/elsewhere/myapp",
-			"/opt/elsewhere/myapp/docker-compose.yml",
-			true,
-		)),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Len(t, jobs, 1)
-	require.True(t, jobs[0].Refused)
-}
-
-func TestFindJobs_MultipleStacksSameImage(t *testing.T) {
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
-			"myapp-prod", "web",
-			"/home/shcizo/self-hosted/myapp-prod",
-			"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
-			true,
-		)),
-		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
-			"myapp-staging", "web",
-			"/home/shcizo/self-hosted/myapp-staging",
-			"/home/shcizo/self-hosted/myapp-staging/docker-compose.yml",
-			true,
-		)),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Len(t, jobs, 2)
-}
-
-func TestFindJobs_DockerError(t *testing.T) {
-	fake := &fakeDockerClient{err: errors.New("connection refused")}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	_, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.Error(t, err)
-}
-
-func TestFindJobs_NonComposeContainerIsSkipped(t *testing.T) {
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", map[string]string{
-			"se.shcizo.auto-update": "true",
-		}),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Empty(t, jobs)
-}
-
-func TestFindJobs_PopulatesImageField(t *testing.T) {
-	fake := &fakeDockerClient{containers: []types.Container{
-		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
-			"myapp-prod", "web",
-			"/home/shcizo/self-hosted/myapp-prod",
-			"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
-			true,
-		)),
-	}}
-	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Len(t, jobs, 1)
-	require.Equal(t, "registry.example.com/myapp", jobs[0].Image)
-	require.Empty(t, jobs[0].ServiceID)
-}

internal/discovery/docker_client.go

@@ -1,15 +0,0 @@
-package discovery
-
-import (
-	"context"
-
-	"github.com/docker/docker/api/types"
-	"github.com/docker/docker/api/types/container"
-)
-
-// DockerClient is the subset of the Docker SDK we depend on.
-// Defined as an interface so tests can supply a fake.
-type DockerClient interface {
-	ContainerList(ctx context.Context, opts container.ListOptions) ([]types.Container, error)
-	Ping(ctx context.Context) (types.Ping, error)
-}

internal/discovery/labels.go

@@ -1,71 +0,0 @@
-package discovery
-
-import (
-	"fmt"
-	"strings"
-)
-
-// ComposeLabels captures the four Compose-managed labels we need
-// to drive a `docker compose pull`/`up -d` against the right stack.
-type ComposeLabels struct {
-	Project     string
-	Service     string
-	WorkingDir  string
-	ConfigFiles []string
-}
-
-const (
-	labelProject     = "com.docker.compose.project"
-	labelService     = "com.docker.compose.service"
-	labelWorkingDir  = "com.docker.compose.project.working_dir"
-	labelConfigFiles = "com.docker.compose.project.config_files"
-)
-
-// ParseComposeLabels extracts the Compose labels we need. Returns an
-// error naming the missing label if any required field is absent —
-// in practice this should only happen if the container was not
-// started by Compose.
-func ParseComposeLabels(labels map[string]string) (ComposeLabels, error) {
-	get := func(key string) (string, error) {
-		v, ok := labels[key]
-		if !ok || v == "" {
-			return "", fmt.Errorf("missing required label: %s", key)
-		}
-		return v, nil
-	}
-
-	project, err := get(labelProject)
-	if err != nil {
-		return ComposeLabels{}, err
-	}
-	service, err := get(labelService)
-	if err != nil {
-		return ComposeLabels{}, err
-	}
-	workingDir, err := get(labelWorkingDir)
-	if err != nil {
-		return ComposeLabels{}, err
-	}
-	configFilesRaw, err := get(labelConfigFiles)
-	if err != nil {
-		return ComposeLabels{}, err
-	}
-
-	files := strings.Split(configFilesRaw, ",")
-	for i, f := range files {
-		files[i] = strings.TrimSpace(f)
-	}
-
-	return ComposeLabels{
-		Project:     project,
-		Service:     service,
-		WorkingDir:  workingDir,
-		ConfigFiles: files,
-	}, nil
-}
-
-// HasOptIn reports whether the labels include the opt-in marker with
-// value "true" (exact, case-sensitive — anything else is excluded).
-func HasOptIn(labels map[string]string, key string) bool {
-	return labels[key] == "true"
-}

internal/discovery/labels_test.go

@@ -1,88 +0,0 @@
-package discovery_test
-
-import (
-	"testing"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/stretchr/testify/require"
-)
-
-func TestParseComposeLabels_Success(t *testing.T) {
-	labels := map[string]string{
-		"com.docker.compose.project":                  "myapp-prod",
-		"com.docker.compose.service":                  "web",
-		"com.docker.compose.project.working_dir":      "/home/shcizo/self-hosted/myapp-prod",
-		"com.docker.compose.project.config_files":     "/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
-		"se.shcizo.auto-update":                       "true",
-	}
-	got, err := discovery.ParseComposeLabels(labels)
-	require.NoError(t, err)
-	require.Equal(t, "myapp-prod", got.Project)
-	require.Equal(t, "web", got.Service)
-	require.Equal(t, "/home/shcizo/self-hosted/myapp-prod", got.WorkingDir)
-	require.Equal(t, []string{"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml"}, got.ConfigFiles)
-}
-
-func TestParseComposeLabels_MultipleConfigFiles(t *testing.T) {
-	labels := map[string]string{
-		"com.docker.compose.project":                  "myapp",
-		"com.docker.compose.service":                  "web",
-		"com.docker.compose.project.working_dir":      "/srv/myapp",
-		"com.docker.compose.project.config_files":     "/srv/myapp/docker-compose.yml,/srv/myapp/docker-compose.prod.yml",
-	}
-	got, err := discovery.ParseComposeLabels(labels)
-	require.NoError(t, err)
-	require.Equal(t, []string{
-		"/srv/myapp/docker-compose.yml",
-		"/srv/myapp/docker-compose.prod.yml",
-	}, got.ConfigFiles)
-}
-
-func TestParseComposeLabels_MissingFieldsError(t *testing.T) {
-	cases := []struct {
-		name    string
-		missing string
-		labels  map[string]string
-	}{
-		{"project", "com.docker.compose.project", map[string]string{
-			"com.docker.compose.service":              "web",
-			"com.docker.compose.project.working_dir":  "/x",
-			"com.docker.compose.project.config_files": "/x/y.yml",
-		}},
-		{"service", "com.docker.compose.service", map[string]string{
-			"com.docker.compose.project":              "p",
-			"com.docker.compose.project.working_dir":  "/x",
-			"com.docker.compose.project.config_files": "/x/y.yml",
-		}},
-		{"working_dir", "com.docker.compose.project.working_dir", map[string]string{
-			"com.docker.compose.project":              "p",
-			"com.docker.compose.service":              "web",
-			"com.docker.compose.project.config_files": "/x/y.yml",
-		}},
-		{"config_files", "com.docker.compose.project.config_files", map[string]string{
-			"com.docker.compose.project":             "p",
-			"com.docker.compose.service":             "web",
-			"com.docker.compose.project.working_dir": "/x",
-		}},
-	}
-	for _, c := range cases {
-		t.Run(c.name, func(t *testing.T) {
-			_, err := discovery.ParseComposeLabels(c.labels)
-			require.Error(t, err)
-			require.Contains(t, err.Error(), c.missing)
-		})
-	}
-}
-
-func TestHasOptIn(t *testing.T) {
-	require.True(t, discovery.HasOptIn(map[string]string{
-		"se.shcizo.auto-update": "true",
-	}, "se.shcizo.auto-update"))
-	require.False(t, discovery.HasOptIn(map[string]string{
-		"se.shcizo.auto-update": "false",
-	}, "se.shcizo.auto-update"))
-	require.False(t, discovery.HasOptIn(map[string]string{
-		"se.shcizo.auto-update": "TRUE",
-	}, "se.shcizo.auto-update"))
-	require.False(t, discovery.HasOptIn(map[string]string{}, "se.shcizo.auto-update"))
-}

internal/discovery/matching.go

@@ -1,33 +0,0 @@
-package discovery
-
-import "strings"
-
-// NormaliseImage strips the tag and digest from an image reference,
-// returning the bare repository name.
-//
-// Tricky case: "localhost:5000/foo:v1" — the first colon is a port,
-// not a tag. We disambiguate by splitting on "/" first and only
-// treating colons in the last segment as tag separators.
-func NormaliseImage(ref string) string {
-	if at := strings.Index(ref, "@"); at >= 0 {
-		ref = ref[:at]
-	}
-	slash := strings.LastIndex(ref, "/")
-	if slash < 0 {
-		if colon := strings.Index(ref, ":"); colon >= 0 {
-			return ref[:colon]
-		}
-		return ref
-	}
-	prefix, last := ref[:slash], ref[slash+1:]
-	if colon := strings.Index(last, ":"); colon >= 0 {
-		last = last[:colon]
-	}
-	return prefix + "/" + last
-}
-
-// ImagesMatch reports whether two image references resolve to the same
-// repository, ignoring tag and digest. Case-sensitive per spec section 5.2.
-func ImagesMatch(a, b string) bool {
-	return NormaliseImage(a) == NormaliseImage(b)
-}

internal/discovery/matching_test.go

@@ -1,51 +0,0 @@
-package discovery_test
-
-import (
-	"testing"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/stretchr/testify/require"
-)
-
-func TestNormaliseImage(t *testing.T) {
-	cases := []struct {
-		in   string
-		want string
-	}{
-		{"registry.example.com/myapp", "registry.example.com/myapp"},
-		{"registry.example.com/myapp:v1.2.3", "registry.example.com/myapp"},
-		{"registry.example.com/myapp:latest", "registry.example.com/myapp"},
-		{"registry.example.com/myapp@sha256:abc123", "registry.example.com/myapp"},
-		{"registry.example.com/myapp:v1.2.3@sha256:abc123", "registry.example.com/myapp"},
-		{"nginx", "nginx"},
-		{"nginx:1.25-alpine", "nginx"},
-		{"library/nginx:latest", "library/nginx"},
-		{"gcr.io/proj/svc:tag", "gcr.io/proj/svc"},
-		{"localhost:5000/myimg:v1", "localhost:5000/myimg"},
-	}
-	for _, c := range cases {
-		t.Run(c.in, func(t *testing.T) {
-			require.Equal(t, c.want, discovery.NormaliseImage(c.in))
-		})
-	}
-}
-
-func TestImagesMatch(t *testing.T) {
-	require.True(t, discovery.ImagesMatch(
-		"registry.example.com/myapp",
-		"registry.example.com/myapp:v1.2.3",
-	))
-	require.True(t, discovery.ImagesMatch(
-		"registry.example.com/myapp:v1.0.0",
-		"registry.example.com/myapp:v9.9.9",
-	))
-	require.False(t, discovery.ImagesMatch(
-		"registry.example.com/myapp",
-		"registry.example.com/otherapp",
-	))
-	// Case-sensitive per spec section 5.2
-	require.False(t, discovery.ImagesMatch(
-		"registry.example.com/MyApp",
-		"registry.example.com/myapp",
-	))
-}

internal/discovery/swarm.go

@@ -1,60 +0,0 @@
-package discovery
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/docker/docker/api/types/swarm"
-)
-
-// SwarmDockerClient is the subset of the Docker SDK SwarmDiscovery depends
-// on. Defined as an interface so tests can supply a fake.
-type SwarmDockerClient interface {
-	ServiceList(ctx context.Context, opts swarm.ServiceListOptions) ([]swarm.Service, error)
-}
-
-// SwarmDiscovery orchestrates "given an image, which Swarm services should
-// we update?". Unlike Discovery, it has no filesystem path to check — the
-// opt-in label on the service itself is the only gate, since Swarm services
-// have no local compose file to anchor a STACKS_ROOT check against.
-type SwarmDiscovery struct {
-	cli        SwarmDockerClient
-	optInLabel string
-}
-
-// NewSwarm returns a SwarmDiscovery bound to the given Docker client and
-// opt-in label.
-func NewSwarm(cli SwarmDockerClient, optInLabel string) *SwarmDiscovery {
-	return &SwarmDiscovery{cli: cli, optInLabel: optInLabel}
-}
-
-// FindJobs lists Swarm services, filters by image match + opt-in label, and
-// returns one Job per matching service. Signature matches Discovery.FindJobs
-// so both satisfy api.Finder.
-func (d *SwarmDiscovery) FindJobs(ctx context.Context, image string) ([]Job, error) {
-	services, err := d.cli.ServiceList(ctx, swarm.ServiceListOptions{})
-	if err != nil {
-		return nil, fmt.Errorf("docker service list: %w", err)
-	}
-
-	var jobs []Job
-	for _, svc := range services {
-		if svc.Spec.TaskTemplate.ContainerSpec == nil {
-			continue
-		}
-		if !ImagesMatch(image, svc.Spec.TaskTemplate.ContainerSpec.Image) {
-			continue
-		}
-		if !HasOptIn(svc.Spec.Labels, d.optInLabel) {
-			continue
-		}
-
-		jobs = append(jobs, Job{
-			Service:   svc.Spec.Name,
-			ServiceID: svc.ID,
-			Image:     image,
-		})
-	}
-
-	return jobs, nil
-}

internal/discovery/swarm_test.go

@@ -1,86 +0,0 @@
-package discovery_test
-
-import (
-	"context"
-	"errors"
-	"testing"
-
-	"github.com/docker/docker/api/types/swarm"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/stretchr/testify/require"
-)
-
-type fakeSwarmDockerClient struct {
-	services []swarm.Service
-	err      error
-}
-
-func (f *fakeSwarmDockerClient) ServiceList(_ context.Context, _ swarm.ServiceListOptions) ([]swarm.Service, error) {
-	return f.services, f.err
-}
-
-func mkService(id, name, image string, labels map[string]string) swarm.Service {
-	return swarm.Service{
-		ID: id,
-		Spec: swarm.ServiceSpec{
-			Annotations: swarm.Annotations{Name: name, Labels: labels},
-			TaskTemplate: swarm.TaskSpec{
-				ContainerSpec: &swarm.ContainerSpec{Image: image},
-			},
-		},
-	}
-}
-
-func TestSwarmFindJobs_MatchAndOptIn(t *testing.T) {
-	fake := &fakeSwarmDockerClient{services: []swarm.Service{
-		mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", map[string]string{
-			"se.shcizo.auto-update": "true",
-		}),
-		mkService("svc-other", "other_web", "registry.example.com/other:v1", map[string]string{
-			"se.shcizo.auto-update": "true",
-		}),
-	}}
-	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Len(t, jobs, 1)
-	require.Equal(t, "myapp_web", jobs[0].Service)
-	require.Equal(t, "svc-myapp", jobs[0].ServiceID)
-	require.Equal(t, "registry.example.com/myapp", jobs[0].Image)
-	require.False(t, jobs[0].Refused)
-	require.Empty(t, jobs[0].WorkingDir)
-	require.Empty(t, jobs[0].ConfigFiles)
-}
-
-func TestSwarmFindJobs_SkipsWithoutOptIn(t *testing.T) {
-	fake := &fakeSwarmDockerClient{services: []swarm.Service{
-		mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", nil),
-	}}
-	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Empty(t, jobs)
-}
-
-func TestSwarmFindJobs_DockerError(t *testing.T) {
-	fake := &fakeSwarmDockerClient{err: errors.New("connection refused")}
-	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
-	_, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.Error(t, err)
-}
-
-func TestSwarmFindJobs_NoContainerSpecIsSkipped(t *testing.T) {
-	fake := &fakeSwarmDockerClient{services: []swarm.Service{
-		{
-			ID: "svc-weird",
-			Spec: swarm.ServiceSpec{
-				Annotations:  swarm.Annotations{Name: "weird", Labels: map[string]string{"se.shcizo.auto-update": "true"}},
-				TaskTemplate: swarm.TaskSpec{ContainerSpec: nil},
-			},
-		},
-	}}
-	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
-	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
-	require.NoError(t, err)
-	require.Empty(t, jobs)
-}

internal/metrics/metrics.go

@@ -1,69 +0,0 @@
-// Package metrics defines and registers the Prometheus collectors
-// exported on /metrics.
-package metrics
-
-import (
-	"net/http"
-
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-)
-
-// Metrics bundles all collectors. Each field is the public handle the
-// rest of the service uses to record observations.
-type Metrics struct {
-	BuildInfo      *prometheus.GaugeVec
-	HTTPRequests   *prometheus.CounterVec
-	UpdateJobs     *prometheus.CounterVec
-	UpdateDuration *prometheus.HistogramVec
-	QueueDepth     prometheus.Gauge
-	LastUpdateTime *prometheus.GaugeVec
-	DockerPingUp   prometheus.Gauge
-}
-
-// New constructs Metrics and registers them with the given registry.
-// Use prometheus.NewRegistry() in tests so collectors don't leak between
-// runs; production code uses prometheus.DefaultRegisterer.
-func New(reg prometheus.Registerer) *Metrics {
-	m := &Metrics{
-		BuildInfo: prometheus.NewGaugeVec(prometheus.GaugeOpts{
-			Name: "package_updater_build_info",
-			Help: "Always 1. Labels carry version/commit for dashboards.",
-		}, []string{"version", "commit"}),
-		HTTPRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
-			Name: "package_updater_http_requests_total",
-			Help: "HTTP requests handled, labelled by endpoint and status.",
-		}, []string{"endpoint", "status_code"}),
-		UpdateJobs: prometheus.NewCounterVec(prometheus.CounterOpts{
-			Name: "package_updater_update_jobs_total",
-			Help: "Update jobs executed, labelled by project/service/status.",
-		}, []string{"project", "service", "status"}),
-		UpdateDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
-			Name:    "package_updater_update_duration_seconds",
-			Help:    "Time taken to pull + up a single service.",
-			Buckets: []float64{0.5, 1, 2, 5, 10, 30, 60, 120, 300},
-		}, []string{"project", "service"}),
-		QueueDepth: prometheus.NewGauge(prometheus.GaugeOpts{
-			Name: "package_updater_queue_depth",
-			Help: "Current number of submissions waiting in the queue.",
-		}),
-		LastUpdateTime: prometheus.NewGaugeVec(prometheus.GaugeOpts{
-			Name: "package_updater_last_update_timestamp",
-			Help: "Unix timestamp of the most recent successful update per service.",
-		}, []string{"project", "service"}),
-		DockerPingUp: prometheus.NewGauge(prometheus.GaugeOpts{
-			Name: "package_updater_docker_ping_up",
-			Help: "1 if the Docker socket responded to ping, 0 otherwise.",
-		}),
-	}
-	reg.MustRegister(
-		m.BuildInfo, m.HTTPRequests, m.UpdateJobs, m.UpdateDuration,
-		m.QueueDepth, m.LastUpdateTime, m.DockerPingUp,
-	)
-	return m
-}
-
-// Handler returns the /metrics HTTP handler bound to the given registry.
-func Handler(gatherer prometheus.Gatherer) http.Handler {
-	return promhttp.HandlerFor(gatherer, promhttp.HandlerOpts{})
-}

internal/selfupdate/selfupdate.go

@@ -1,56 +0,0 @@
-// Package selfupdate handles the special case where the update target
-// is the running service's own container. We must finish writing the
-// HTTP response (and flush + close the connection) before exec'ing
-// `docker compose up -d` against ourselves, otherwise the response is
-// lost when the container is replaced.
-package selfupdate
-
-import (
-	"context"
-	"time"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-)
-
-// IsSelf reports whether job targets the running service.
-// Matches on Compose project name (which is also typically the
-// service name for single-service stacks).
-func IsSelf(job discovery.Job, selfProject string) bool {
-	return job.Project == selfProject
-}
-
-// innerExec is the executor abstraction we wrap.
-type innerExec interface {
-	Execute(ctx context.Context, job discovery.Job) error
-}
-
-// Wrapped wraps an Executor with self-update-aware deferred execution.
-type Wrapped struct {
-	inner       innerExec
-	selfProject string
-	delay       time.Duration
-}
-
-// Wrap returns a Wrapped that defers exec until after flush() for
-// self-updates. The delay is added after flush before exec, so the
-// kernel TCP buffer has time to drain.
-func Wrap(inner innerExec, selfProject string, delay time.Duration) *Wrapped {
-	return &Wrapped{inner: inner, selfProject: selfProject, delay: delay}
-}
-
-// ExecuteWithFlush runs job, invoking flush() before exec for self-updates
-// and waiting `delay` after flush. For non-self jobs, exec happens
-// normally and flush is not invoked at all (the HTTP layer decides
-// when to flush in that case).
-func (w *Wrapped) ExecuteWithFlush(ctx context.Context, job discovery.Job, flush func()) error {
-	if !IsSelf(job, w.selfProject) {
-		return w.inner.Execute(ctx, job)
-	}
-	flush()
-	select {
-	case <-time.After(w.delay):
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-	return w.inner.Execute(ctx, job)
-}

internal/selfupdate/selfupdate_test.go

@@ -1,87 +0,0 @@
-package selfupdate_test
-
-import (
-	"context"
-	"errors"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/selfupdate"
-	"github.com/stretchr/testify/require"
-)
-
-type recExec struct {
-	mu     sync.Mutex
-	called bool
-	at     time.Time
-}
-
-func (r *recExec) Execute(_ context.Context, _ discovery.Job) error {
-	r.mu.Lock()
-	r.called = true
-	r.at = time.Now()
-	r.mu.Unlock()
-	return nil
-}
-
-func TestIsSelf(t *testing.T) {
-	require.True(t, selfupdate.IsSelf(
-		discovery.Job{Project: "package-updater", Service: "package-updater"},
-		"package-updater",
-	))
-	require.False(t, selfupdate.IsSelf(
-		discovery.Job{Project: "other", Service: "web"},
-		"package-updater",
-	))
-}
-
-func TestWrap_DefersSelf(t *testing.T) {
-	inner := &recExec{}
-	wrapped := selfupdate.Wrap(inner, "package-updater", 30*time.Millisecond)
-
-	flushed := make(chan time.Time, 1)
-	flush := func() { flushed <- time.Now() }
-
-	job := discovery.Job{Project: "package-updater", Service: "package-updater"}
-	require.NoError(t, wrapped.ExecuteWithFlush(context.Background(), job, flush))
-
-	flushAt := <-flushed
-	inner.mu.Lock()
-	require.True(t, inner.called)
-	require.True(t, inner.at.After(flushAt))
-	inner.mu.Unlock()
-}
-
-func TestWrap_NormalJobIsImmediate(t *testing.T) {
-	inner := &recExec{}
-	wrapped := selfupdate.Wrap(inner, "package-updater", 30*time.Millisecond)
-
-	flushed := make(chan time.Time, 1)
-	flush := func() { flushed <- time.Now() }
-
-	job := discovery.Job{Project: "other", Service: "web"}
-	require.NoError(t, wrapped.ExecuteWithFlush(context.Background(), job, flush))
-
-	inner.mu.Lock()
-	require.True(t, inner.called)
-	inner.mu.Unlock()
-	select {
-	case <-flushed:
-		t.Fatal("flush should not be called for non-self jobs")
-	default:
-	}
-}
-
-func TestWrap_PropagatesError(t *testing.T) {
-	wrapped := selfupdate.Wrap(failExec{}, "x", 1*time.Millisecond)
-	err := wrapped.ExecuteWithFlush(context.Background(), discovery.Job{Project: "other"}, func() {})
-	require.Error(t, err)
-}
-
-type failExec struct{}
-
-func (failExec) Execute(_ context.Context, _ discovery.Job) error {
-	return errors.New("boom")
-}

internal/updater/executor.go

@@ -1,15 +0,0 @@
-// Package updater contains the FIFO job queue, the single worker that
-// drains it, and the abstraction over `docker compose` invocation.
-package updater
-
-import (
-	"context"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-)
-
-// Executor runs `docker compose pull` then `up -d` for a single job.
-// Defined as an interface so the worker can be tested with a fake.
-type Executor interface {
-	Execute(ctx context.Context, job discovery.Job) error
-}

internal/updater/queue.go

@@ -1,144 +0,0 @@
-package updater
-
-import (
-	"context"
-	"time"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/metrics"
-)
-
-// Status describes the outcome of executing a single Job.
-type Status string
-
-const (
-	StatusUpdated Status = "updated"
-	StatusFailed  Status = "failed"
-	StatusRefused Status = "refused"
-	StatusTimeout Status = "timeout"
-)
-
-// Result captures a per-Job outcome to be surfaced in the HTTP response.
-type Result struct {
-	Job        discovery.Job
-	Status     Status
-	Error      string
-	DurationMs int64
-}
-
-// Queue serialises Job execution through a single worker so we never
-// run two `docker compose` commands against the same stack concurrently.
-// It also satisfies the "single global FIFO worker" design choice in
-// spec section 5.7.
-type Queue struct {
-	exec    Executor
-	metrics *metrics.Metrics
-	ch      chan submission
-	stop    chan struct{}
-	done    chan struct{}
-}
-
-type submission struct {
-	ctx     context.Context
-	jobs    []discovery.Job
-	results chan []Result
-}
-
-// NewQueue constructs a queue bound to the given executor. m may be nil,
-// in which case metrics recording is silently skipped.
-func NewQueue(exec Executor, m *metrics.Metrics) *Queue {
-	return &Queue{
-		exec:    exec,
-		metrics: m,
-		ch:      make(chan submission, 16),
-		stop:    make(chan struct{}),
-		done:    make(chan struct{}),
-	}
-}
-
-// Start launches the single background worker. Call Stop to terminate.
-func (q *Queue) Start(ctx context.Context) {
-	go q.run(ctx)
-}
-
-// Stop signals the worker to exit and waits for it to finish.
-func (q *Queue) Stop() {
-	close(q.stop)
-	<-q.done
-}
-
-// Submit blocks until all jobs in the batch have been processed by the
-// worker, then returns per-job results in the same order. The ctx
-// timeout (if any) applies to each individual Job's execution and is
-// surfaced as StatusTimeout.
-func (q *Queue) Submit(ctx context.Context, jobs []discovery.Job) []Result {
-	resCh := make(chan []Result, 1)
-	q.ch <- submission{ctx: ctx, jobs: jobs, results: resCh}
-	if q.metrics != nil {
-		q.metrics.QueueDepth.Set(float64(len(q.ch)))
-	}
-	return <-resCh
-}
-
-func (q *Queue) run(_ context.Context) {
-	defer close(q.done)
-	for {
-		select {
-		case <-q.stop:
-			return
-		case s := <-q.ch:
-			results := make([]Result, len(s.jobs))
-			for i, j := range s.jobs {
-				results[i] = q.runOne(s.ctx, j)
-			}
-			s.results <- results
-		}
-	}
-}
-
-func (q *Queue) runOne(ctx context.Context, job discovery.Job) Result {
-	start := time.Now()
-	r := Result{Job: job}
-
-	if job.Refused {
-		r.Status = StatusRefused
-		r.Error = job.RefusedReason
-		r.DurationMs = time.Since(start).Milliseconds()
-		if q.metrics != nil {
-			q.metrics.UpdateJobs.WithLabelValues(job.Project, job.Service, string(StatusRefused)).Inc()
-			q.metrics.UpdateDuration.WithLabelValues(job.Project, job.Service).Observe(time.Since(start).Seconds())
-		}
-		return r
-	}
-
-	err := q.exec.Execute(ctx, job)
-	r.DurationMs = time.Since(start).Milliseconds()
-	switch {
-	case err == nil:
-		r.Status = StatusUpdated
-	case errorsIsContextDeadline(err) || ctxDeadlineExceeded(ctx):
-		r.Status = StatusTimeout
-		r.Error = err.Error()
-	default:
-		r.Status = StatusFailed
-		r.Error = err.Error()
-	}
-
-	if q.metrics != nil {
-		elapsed := time.Since(start).Seconds()
-		q.metrics.UpdateJobs.WithLabelValues(job.Project, job.Service, string(r.Status)).Inc()
-		q.metrics.UpdateDuration.WithLabelValues(job.Project, job.Service).Observe(elapsed)
-		if r.Status == StatusUpdated {
-			q.metrics.LastUpdateTime.WithLabelValues(job.Project, job.Service).SetToCurrentTime()
-		}
-	}
-	return r
-}
-
-func errorsIsContextDeadline(err error) bool {
-	return err == context.DeadlineExceeded
-}
-
-func ctxDeadlineExceeded(ctx context.Context) bool {
-	return ctx.Err() == context.DeadlineExceeded
-}

internal/updater/queue_test.go

@@ -1,129 +0,0 @@
-package updater_test
-
-import (
-	"context"
-	"errors"
-	"sync"
-	"testing"
-	"time"
-
-	dto "github.com/prometheus/client_model/go"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/metrics"
-	"github.com/shcizo/package-updater/internal/updater"
-	"github.com/stretchr/testify/require"
-)
-
-type fakeExec struct {
-	mu        sync.Mutex
-	calls     []discovery.Job
-	delay     time.Duration
-	errForSvc map[string]error
-}
-
-func (f *fakeExec) Execute(ctx context.Context, job discovery.Job) error {
-	f.mu.Lock()
-	f.calls = append(f.calls, job)
-	f.mu.Unlock()
-	if f.delay > 0 {
-		select {
-		case <-time.After(f.delay):
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-	if err, ok := f.errForSvc[job.Service]; ok {
-		return err
-	}
-	return nil
-}
-
-func (f *fakeExec) callsCopy() []discovery.Job {
-	f.mu.Lock()
-	defer f.mu.Unlock()
-	return append([]discovery.Job(nil), f.calls...)
-}
-
-func TestQueue_ProcessesFIFO(t *testing.T) {
-	exec := &fakeExec{delay: 20 * time.Millisecond}
-	q := updater.NewQueue(exec, nil)
-	q.Start(context.Background())
-	defer q.Stop()
-
-	job1 := discovery.Job{Project: "a", Service: "svc"}
-	job2 := discovery.Job{Project: "b", Service: "svc"}
-	job3 := discovery.Job{Project: "c", Service: "svc"}
-
-	results := make(chan []updater.Result, 3)
-	go func() { results <- q.Submit(context.Background(), []discovery.Job{job1}) }()
-	time.Sleep(5 * time.Millisecond)
-	go func() { results <- q.Submit(context.Background(), []discovery.Job{job2}) }()
-	time.Sleep(5 * time.Millisecond)
-	go func() { results <- q.Submit(context.Background(), []discovery.Job{job3}) }()
-
-	for i := 0; i < 3; i++ {
-		<-results
-	}
-
-	calls := exec.callsCopy()
-	require.Equal(t, []string{"a", "b", "c"}, []string{calls[0].Project, calls[1].Project, calls[2].Project})
-}
-
-func TestQueue_ReturnsPerJobResults(t *testing.T) {
-	exec := &fakeExec{errForSvc: map[string]error{"failing": errors.New("boom")}}
-	q := updater.NewQueue(exec, nil)
-	q.Start(context.Background())
-	defer q.Stop()
-
-	jobs := []discovery.Job{
-		{Project: "p1", Service: "ok"},
-		{Project: "p2", Service: "failing"},
-		{Project: "p3", Service: "refused-svc", Refused: true, RefusedReason: "outside root"},
-	}
-	results := q.Submit(context.Background(), jobs)
-
-	require.Len(t, results, 3)
-	require.Equal(t, updater.StatusUpdated, results[0].Status)
-	require.Equal(t, updater.StatusFailed, results[1].Status)
-	require.Contains(t, results[1].Error, "boom")
-	require.Equal(t, updater.StatusRefused, results[2].Status)
-	require.Contains(t, results[2].Error, "outside root")
-}
-
-func TestQueue_Timeout(t *testing.T) {
-	exec := &fakeExec{delay: 200 * time.Millisecond}
-	q := updater.NewQueue(exec, nil)
-	q.Start(context.Background())
-	defer q.Stop()
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
-	defer cancel()
-
-	jobs := []discovery.Job{{Project: "slow", Service: "svc"}}
-	results := q.Submit(ctx, jobs)
-
-	require.Len(t, results, 1)
-	require.Equal(t, updater.StatusTimeout, results[0].Status)
-}
-
-func TestQueue_RecordsMetrics(t *testing.T) {
-	reg := prometheus.NewRegistry()
-	m := metrics.New(reg)
-	exec := &fakeExec{}
-	q := updater.NewQueue(exec, m)
-	q.Start(context.Background())
-	defer q.Stop()
-
-	q.Submit(context.Background(), []discovery.Job{{Project: "p", Service: "s"}})
-
-	// Verify UpdateJobs counter was incremented for the successful job.
-	metric := &dto.Metric{}
-	require.NoError(t, m.UpdateJobs.WithLabelValues("p", "s", "updated").Write(metric))
-	require.Equal(t, 1.0, metric.GetCounter().GetValue())
-
-	// Verify LastUpdateTime was set (non-zero).
-	tsMetric := &dto.Metric{}
-	require.NoError(t, m.LastUpdateTime.WithLabelValues("p", "s").Write(tsMetric))
-	require.Greater(t, tsMetric.GetGauge().GetValue(), 0.0)
-}

internal/updater/worker.go

@@ -1,6 +0,0 @@
-package updater
-
-// Worker logic currently lives in queue.go (single in-process worker
-// goroutine). This file is reserved for the eventual per-stack-mutex
-// implementation called out in spec section 15 ("Multi-worker concurrency
-// with per-stack mutex"). Intentionally empty for v1.

internal/updater/worker_test.go

@@ -1,38 +0,0 @@
-package updater_test
-
-import (
-	"context"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/shcizo/package-updater/internal/discovery"
-	"github.com/shcizo/package-updater/internal/updater"
-	"github.com/stretchr/testify/require"
-)
-
-type counterExec struct{ n atomic.Int32 }
-
-func (c *counterExec) Execute(_ context.Context, _ discovery.Job) error {
-	c.n.Add(1)
-	time.Sleep(10 * time.Millisecond)
-	return nil
-}
-
-func TestWorker_RunsExactlyOneAtATime(t *testing.T) {
-	exec := &counterExec{}
-	q := updater.NewQueue(exec, nil)
-	q.Start(context.Background())
-	defer q.Stop()
-
-	jobs := make([]discovery.Job, 10)
-	for i := range jobs {
-		jobs[i] = discovery.Job{Project: "p", Service: "svc"}
-	}
-	start := time.Now()
-	q.Submit(context.Background(), jobs)
-	elapsed := time.Since(start)
-
-	require.GreaterOrEqual(t, elapsed, 90*time.Millisecond)
-	require.Equal(t, int32(10), exec.n.Load())
-}