diff --git a/CLAUDE.md b/CLAUDE.md index 47e7d27..bf4e34b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,7 +22,7 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ ## Architecture (one-line per package) -- `cmd/master` — wiring only: config → store → hub → jobs → web, two listeners (UI, agents over pinned TLS) +- `cmd/master` — wiring only: config → store → hub → jobs → web + webhook, two listeners (UI, agents over pinned TLS) - `cmd/agent` — wiring only: config → docker client → inventory → job runner → master connection - `internal/protocol` — the WebSocket messages both sides speak (`Version`, `JobStart`, `JobResult`, ...) - `internal/pin` — master key pair, certificate and fingerprint pinning @@ -37,6 +37,7 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ - `internal/master/store` — Postgres access and migrations - `internal/master/hub` — agent connections: auth, per-connection writer, routing, job reconcile on hello - `internal/master/jobs` — decides whether a service may be updated and starts a job on the right agent +- `internal/master/webhook` — `POST /update`: records the available version, starts jobs for auto services, waits for them - `internal/master/web`, `webauth` — HTMX UI behind OIDC login, SSE signals - `internal/master/events` — in-process publish/subscribe for UI refresh signals - `internal/master/model` — domain types shared by the above @@ -59,8 +60,11 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ - **Unknown message types are logged and ignored; malformed messages close the connection.** - **Reconcile before routable**: the hub fails a reconnecting agent's jobs it does not list in `known_jobs`, and only then accepts jobs for that connection. -- **Auth**: agents by token over pinned TLS; people by OIDC. Every state-changing UI - request is a POST with a CSRF token. +- **The webhook never creates services; new services always start `manual`.** The upsert + does not touch `update_mode`. +- **Auth**: agents by token over pinned TLS; people by OIDC; CI by the `UPDATER_API_KEY` + bearer token on `POST /update` only. Every state-changing UI request is a POST with a + CSRF token. ## Gotchas @@ -89,7 +93,18 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ - **Swarm "succeeded" means the daemon accepted the new spec**, not that tasks converged. - **Swarm updates need manager-node API access** for the agent doing the update. - **The image's default entrypoint is `updater-agent`.** The master overrides it with - `updater-master`. There is no `POST /update` until phase 3. + `updater-master`. +- **The webhook waits via a broker signal plus a store re-read, with a fallback poll** + (the broker drops signals for slow subscribers). Marking and starting jobs runs on + `context.WithoutCancel`, so a client that disconnects cannot leave a queued job no agent + knows about; only the waiting stops. The master never fails or cancels a timed-out job. +- **A webhook job's target is `:`**, built from the service's own + repo, never from the request's image string. With an explicit target the "follows no tag" + refusal does not apply; removed, self, built and container still do. +- **`available_*` is cleared by `FinishJob`** on success only when the job's `to_ref` + (as created, without `@digest`) equals `available_ref` and `available_at <= job.created_at` + (that version is now running). A manual Update of the followed tag does not clear a + badge for a different tag. - **Job output is capped at `protocol.MaxJobOutput` (128 KiB, tail).** NUL bytes are stripped by the hub because Postgres text rejects them. @@ -98,10 +113,10 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ - Master/agent spec: `docs/superpowers/specs/2026-10-12-master-agent-design.md` (authoritative) - Phase 1 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-1-inventory.md` - Phase 2 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-2-jobs.md` -- Original single-instance design (historical, its webhook is gone until phase 3): +- Phase 3 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md` +- Original single-instance design (historical): `docs/superpowers/specs/2026-05-22-package-updater-design.md` -- Consumer-side CI integration: `gitea-action/` (calls `POST /update`, which no longer - exists until phase 3) +- Consumer-side CI integration: `gitea-action/` (calls the master's `POST /update`) ## Conventions diff --git a/README.md b/README.md index 5c26107..dc79ae5 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,7 @@ the same image: `updater-master` and `updater-agent` (the default entrypoint). Design: [master/agent spec](docs/superpowers/specs/2026-10-12-master-agent-design.md). Phase 2 plan: [jobs and manual update](docs/superpowers/plans/2026-10-12-master-agent-phase-2-jobs.md). - -> **Webhook:** the old `POST /update` endpoint is gone. It returns in phase 3 on the -> master; until then the Gitea action in `gitea-action/` has nothing to call. +Phase 3 plan: [webhook and auto mode](docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md). ## What an update does @@ -20,9 +18,9 @@ runs is exactly what was pulled. The log streams to the browser while it runs; j history is kept in Postgres. A service without a tag (`repo@digest`), a plain container, and the updater's own services cannot be updated, and the UI says why. A bare image (`image: postgres`, no tag and no digest) counts as `postgres:latest`; only a reference -with a digest and no tag (`repo@sha256:...`) "follows no tag". Compose services with a `build:` section are shown but never updated -("built from source (build:)"). One update at a -time per service, and one job at a time per agent. +with a digest and no tag (`repo@sha256:...`) "follows no tag". Compose services with a +`build:` section are shown but never updated ("built from source (build:)"). One update at +a time per service, and one job at a time per agent. ## Deploying the master @@ -31,6 +29,89 @@ needs Postgres and an OIDC provider). The master listens for the UI on `PORT` (d `8080`, put it behind your reverse proxy) and for agents on `AGENT_PORT` (default `8090`, pinned TLS, reached directly on the internal network). +| Variable | Required | Default | Purpose | +|---|---|---|---| +| `UPDATER_API_KEY` | no | — | Bearer token for `POST /update`. Without it the webhook answers `503` and the master logs `webhook_disabled`. | +| `UPDATE_TIMEOUT` | no | `5m` | How long one `POST /update` waits for its auto jobs (Go duration). | + +The master's other settings (`DATABASE_URL`, `PUBLIC_URL`, `AGENT_URL`, `OIDC_*`, +`ALLOWED_GROUP`, `AGENT_IMAGE`, `PORT`, `AGENT_PORT`, `LOG_LEVEL`) are shown in +[`deploy/master.stack.yml`](deploy/master.stack.yml). + +## Webhook (CI) + +CI calls the master after pushing an image. The Gitea action in +[`gitea-action/`](gitea-action/README.md) does this; any HTTP client works. + +``` +POST /update +Authorization: Bearer +Content-Type: application/json + +{"image": "registry.example.com/myapp", "tag": "1.5.0"} +``` + +`image` is matched against the image repo of every service on every agent (tag and digest +in the request are ignored for matching; matching is case-sensitive). `tag` is optional. +For every match the master records the version as *available* (the overview shows an +"update available" badge), including services that cannot be updated automatically +(updater's own, `build:` and plain containers), so the badge shows for them too. Services +in **manual** mode stop there. Services in **auto** +mode also get an update job, and the request waits for those jobs to finish, at most +`UPDATE_TIMEOUT`. + +**`tag` semantics.** The target of an auto job is `:`, built +from the service's own repo and never from the request's `image` string. The tag must be +one that has been pushed. Compose services get pinned to it through +`compose.override.yaml`, as in any update. Omit `tag` to re-pull the tag the service +already follows (a bare image counts as `:latest`); a service that follows no tag +(`repo@digest`) is then not touched and is reported as `refused` ("no tag given and the +service follows no tag"), manual or auto. +With a `tag`, the "follows no tag" refusal does not apply; the other refusals (the +updater's own services, `build:` services, plain containers) still do. + +Response (`200` or `500`): + +```json +{"request_id": "…", "image": "registry.example.com/myapp", "tag": "1.5.0", "matched": 2, + "results": [ + {"service_id": 12, "service": "myapp/web", "kind": "compose", "mode": "auto", + "status": "succeeded", "job_id": 42, "job_url": "https://updater.example.com/jobs/42", + "to_ref": "registry.example.com/myapp:1.5.0", "duration_ms": 8210}, + {"service_id": 15, "service": "myapp/worker", "kind": "compose", "mode": "manual", + "status": "marked", "to_ref": "registry.example.com/myapp:1.5.0", "duration_ms": 3}]} +``` + +| `status` | Meaning | Counts as failure | +|---|---|---| +| `succeeded` | Auto job finished successfully. | no | +| `failed` | Auto job failed (for example agent offline), or the version could not be recorded. `error` says why. | yes | +| `timeout` | Job still running when `UPDATE_TIMEOUT` passed. The master leaves it running. | yes | +| `marked` | Manual service; the version was recorded only. | no | +| `busy` | The service already has a queued or running job; `job_id` is that job, which is not waited for. | no | +| `refused` | The service may not be updated, or there is no target (no `tag` given and the service follows no tag). `error` has the reason. | no | + +| Code | When | +|---|---| +| `200` | No result is `failed` or `timeout`. | +| `400` | Body is not JSON, `image` is missing, or `tag` is not a valid Docker tag. | +| `401` | Missing or wrong bearer token. | +| `404` | No service uses the image (catches typos in CI configuration). | +| `500` | At least one result is `failed` or `timeout`. | +| `503` | `UPDATER_API_KEY` is not set on the master. | + +The webhook never creates services: the first deployment of a stack is done by hand. A +new service always starts in **manual** mode; switch it with the *Make auto* button on the +overview or the service page. Auto is refused for the updater's own services, `build:` +services and plain containers. + +> **Warning:** auto mode deploys whatever CI pushes, without anyone looking at it. Using +> it on critical services is your own risk. + +The request is open for as long as the jobs run, so the reverse proxy in front of the +master must allow a read timeout longer than `UPDATE_TIMEOUT`. If the client disconnects, +marking and starting jobs still complete; only the waiting stops. + ## Adding agents Open the master's **Agents** page and add one. It shows a ready-to-run compose file with @@ -115,8 +196,9 @@ crash-loop the next time its image is pulled. Anything that pulls `:latest` for instance triggers this: its own `/update` webhook, Watchtower or Shepherd. Pinning means two things: pin the old instance to the last pre-phase-2 tag **and** remove -its `se.shcizo.auto-update` label, so nothing updates it to the new image. Or do not tag a -release until phase 3 moves `POST /update` to the master. +its `se.shcizo.auto-update` label, so nothing updates it to the new image. The master +serves `POST /update` again (see [Webhook](#webhook-ci)); point the Gitea action's +`endpoint` there and set `UPDATER_API_KEY` on the master. Master and agents must be on this release together: a phase-1 agent connects but every job sent to it fails "agent too old for updates". diff --git a/cmd/master/main.go b/cmd/master/main.go index c942ff4..15cde34 100644 --- a/cmd/master/main.go +++ b/cmd/master/main.go @@ -23,6 +23,7 @@ import ( "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/master/webhook" "github.com/shcizo/package-updater/internal/pin" ) @@ -93,6 +94,14 @@ func run() error { } _, _ = w.Write([]byte("ok")) }) + // The webhook does its own bearer-token check, so it sits outside the + // OIDC login like /healthz. + mux.Handle("POST /update", webhook.New(st, jobService, broker, webhook.Config{ + Token: cfg.UpdaterAPIKey, Timeout: cfg.UpdateTimeout, PublicURL: cfg.PublicURL, + }, logger)) + if cfg.UpdaterAPIKey == "" { + logger.Warn("webhook_disabled", "reason", "UPDATER_API_KEY not set") + } 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))) @@ -103,8 +112,8 @@ func run() error { // 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. + // No WriteTimeout on either server: SSE streams, agent connections and + // POST /update (which waits up to UPDATE_TIMEOUT) are long-lived by design. webSrv := &http.Server{ Addr: ":" + cfg.Port, Handler: mux, ReadHeaderTimeout: 10 * time.Second, BaseContext: baseContext, diff --git a/deploy/master.stack.yml b/deploy/master.stack.yml index e4f52e7..3c0e413 100644 --- a/deploy/master.stack.yml +++ b/deploy/master.stack.yml @@ -1,4 +1,4 @@ -# Example Swarm stack for the package-updater master (phase 2: inventory and manual updates). +# Example Swarm stack for the package-updater master (inventory, updates and the CI webhook). # Deploy on the manager: docker stack deploy -c master.stack.yml package-updater-master # # Before deploying: @@ -6,19 +6,25 @@ # 2. In Authentik: create an OAuth2/OpenID provider + application with redirect URI # https://updater.example.com/auth/callback, and a group "updater-admins". # 3. docker secret create pu_oidc_client_secret - ; docker secret create pu_database_url - +# 4. docker secret create pu_updater_api_key - (required by this example; it enables POST /update) +# To run without the webhook, remove the pu_updater_api_key lines (the UPDATER_API_KEY +# part of the entrypoint, the service secret and the secrets entry); the endpoint then answers 503. +# The reverse proxy must allow a read timeout longer than UPDATE_TIMEOUT (default 5m). services: master: image: gitea.shcizo.se/shcizo/package-updater:latest - entrypoint: ["/bin/sh", "-c", "DATABASE_URL=$$(cat /run/secrets/pu_database_url) OIDC_CLIENT_SECRET=$$(cat /run/secrets/pu_oidc_client_secret) exec /usr/local/bin/updater-master"] + entrypoint: ["/bin/sh", "-c", "DATABASE_URL=$$(cat /run/secrets/pu_database_url) OIDC_CLIENT_SECRET=$$(cat /run/secrets/pu_oidc_client_secret) UPDATER_API_KEY=$$(cat /run/secrets/pu_updater_api_key) exec /usr/local/bin/updater-master"] environment: PUBLIC_URL: https://updater.example.com AGENT_URL: wss://10.0.0.5:8090/agent OIDC_ISSUER: https://auth.example.com/application/o/package-updater/ OIDC_CLIENT_ID: package-updater ALLOWED_GROUP: updater-admins + # UPDATE_TIMEOUT: 5m secrets: - pu_database_url - pu_oidc_client_secret + - pu_updater_api_key ports: # Agents connect here directly over the internal network, not via the proxy. - target: 8090 @@ -35,6 +41,8 @@ secrets: external: true pu_oidc_client_secret: external: true + pu_updater_api_key: + external: true networks: proxy: diff --git a/docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md b/docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md new file mode 100644 index 0000000..5bf335a --- /dev/null +++ b/docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md @@ -0,0 +1,148 @@ +# Master/Agent Phase 3 — Webhook and Auto Mode Implementation Plan + +> This plan documents what was built for issue #8 and where it deviates from the spec. Code in "Frozen contracts" is final; tasks give the contract and the test cases. + +**Goal:** CI calls `POST /update` on the master after pushing an image. The master records the pushed version as *available* on every service that uses the image. Services in `auto` mode are updated at once and the request waits for the result, so CI sees a broken deploy. People switch a service between `manual` and `auto` in the UI. + +**Architecture:** A new package `internal/master/webhook` is an `http.Handler` mounted on the master's mux outside the OIDC login; it does its own bearer-token check. It looks up services by image repo, writes `available_*` on each match, starts a job through `jobs.Service.Start` for the `auto` ones and waits for them. Waiting subscribes to the in-process broker before any job exists, reacts to `events.Jobs`, re-reads each pending job from Postgres on every signal and on a fallback ticker, and gives up at `UPDATE_TIMEOUT`. `FinishJob` clears `available_*` in the transaction that records a successful deploy. + +**Tech Stack:** unchanged. No new dependencies. + +**Spec:** `docs/superpowers/specs/2026-10-12-master-agent-design.md`, "Webhook" section. Phase 2 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-2-jobs.md` (its conventions still apply). + +## Global Constraints + +- `go test ./...` passes with no Docker daemon and no Postgres. Postgres tests carry `//go:build integration` and need `DATABASE_URL`. Run `go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/`. +- Interfaces are declared in the consuming package. Hand-written fakes, `stretchr/testify`, table-driven where there is more than one case. +- Every exported identifier gets a GoDoc comment. Conventional Commits; every commit message ends with ` (#8)`. +- The token (`UPDATER_API_KEY`) never appears in logs. +- Every state-changing UI request is a POST carrying the CSRF token (the mode toggle included). +- The HTTP server for the UI has no `WriteTimeout`, so a webhook request can wait up to `UPDATE_TIMEOUT`. + +## Deliberate deviations from the spec in this phase + +- **No `image_versions` table and no `available_digest` column.** Both belong to phase 4 (version history and the version picker). Phase 3 stores only `available_ref`, `available_source` (`webhook`) and `available_at` on `services`. +- **The old `207` is gone.** A request with both a working and a failed service answers `500` as a whole; there is no partial-success code. The Gitea action fails on any 4xx/5xx, so CI surfaces it. +- **`busy` and `refused` are not failures.** A service that already has an active job (`busy`, with that job's ID, not waited for) or may not be updated (`refused`, with the reason) does not make the request fail. Only `failed` and `timeout` do. A manual match is `marked`, also not a failure. +- **A webhook with a `tag` bypasses the "follows no tag" refusal.** The target is then explicit (`:`), so a `repo@digest` service can be moved to a tag. Removed, self, built and container refusals still apply. Without a `tag`, the target is the followed tag (bare image = `:latest`); a service that follows no tag is left alone (`available_*` untouched) and is reported `refused` ("no tag given and the service follows no tag"), manual or auto. Because of this, the UI allows `auto` for a service that follows no tag (`jobs.AutoAllowed`), unlike the Update button (`jobs.Updatable`). +- **The target ref is built from the service's own repo.** `imageref.Parse(req.Image).Repo` only selects services; it never reaches the job, so a request cannot point a service at another registry. +- **Token is optional on the master.** Unset `UPDATER_API_KEY` means `POST /update` answers `503` and startup logs `webhook_disabled`; the master still runs. +- **`tag` is validated** against Docker's tag grammar (`400 invalid tag`). The spec does not say; without it a tag such as `v1@sha256:x` or one with spaces would end up in the override file and fail on the agent. +- **An oversized body (over 64 KiB) answers `400 invalid JSON body`**, not `413`. +- **`UPDATE_TIMEOUT` exists on both sides** with the same name and parsing: on the agent it bounds one job, on the master it bounds one webhook request. The master never fails or cancels a job that outlives the request; the agent's own timeout does. +- **Marking and starting jobs runs on `context.WithoutCancel(r.Context())`.** A client that disconnects stops only the waiting; otherwise a queued job could be left that no agent knows about. + +## Review Focus + +| Failure case | Test that covers it | +|---|---| +| Wrong, missing or malformed token never reaches the lookup; unset key answers 503. | webhook `TestAuth` | +| An unknown or differently-cased image answers 404 and starts nothing. | webhook `TestNoMatch` | +| The image string is normalised (tag, digest, registry with port) before matching. | webhook `TestImageIsNormalisedBeforeMatching` | +| The auto job's target is built from the service's repo and the request tag, never the request's image string. | webhook `TestAutoJobTargetIsBuiltFromServiceRepo` | +| A job that fails at start (agent offline) is reported at once, not waited for. | webhook `TestJobFailedAtStartIsNotWaitedFor` | +| A dropped broker signal is recovered by the fallback poll. | webhook `TestDroppedSignalIsRecoveredByPolling` | +| A timed-out job is reported `timeout` and left untouched. | webhook `TestTimeout` | +| A client that disconnects does not abort marking or starting. | webhook `TestClientCancelStopsWaitingButNotStarting` | +| The webhook waits for the real agent's result and answers 200 / 500. | e2e `TestWebhookWaitsForAutoJob`, `TestWebhookFailedAutoJobIs500` | +| The available ref is cleared only by a successful job whose `to_ref` equals it and that was created at or after the suggestion; a different ref, a newer suggestion and a failed job keep it. | store `TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed` | +| Re-reporting inventory never resets a service's update mode; new services are manual. | store `TestUpsertKeepsUpdateMode` | +| `auto` is refused for removed, self, built and container services; `manual` is always allowed. | jobs `TestAutoAllowed`; web `TestSetMode_AutoRefusedShows409`, `TestSetMode_ManualAlwaysAllowed` | + +## Frozen contracts + +### C1. Migration `internal/master/store/migrations/0003_webhook.sql` (complete) + +```sql +ALTER TABLE services ADD COLUMN available_ref text NOT NULL DEFAULT ''; +ALTER TABLE services ADD COLUMN available_source text NOT NULL DEFAULT '' + CHECK (available_source IN ('', 'webhook')); +ALTER TABLE services ADD COLUMN available_at timestamptz; + +CREATE INDEX services_by_image_repo ON services (image_repo) WHERE removed_at IS NULL; +``` + +### C2. `internal/master/model` additions + +`Service` gains `AvailableRef string`, `AvailableSource string`, `AvailableAt *time.Time`. Constants `ModeManual`, `ModeAuto`, `SourceWebhook`. `type Available struct { Ref, Source string; At time.Time }`. + +### C3. Store methods + +```go +ServicesByRepo(ctx, repo string) ([]model.Service, error) // non-removed, same order as ListServices +SetAvailable(ctx, id int64, a model.Available) error +SetUpdateMode(ctx, id int64, mode string) error // model.ErrNotFound for unknown or removed +``` + +`FinishJob` additionally reads the job's `created_at` and, on success, clears `available_*` in the same transaction when the job's `to_ref` (as created, without `@digest`) equals `available_ref` and `available_at <= created_at`. + +### C4. `internal/master/jobs` + +```go +type StartRequest struct { + Trigger model.JobTrigger + RequestedBy string + Ref string // empty: the service's followed tag; set: used as is +} +func (s *Service) Start(ctx, svc model.Service, req StartRequest) (model.Job, error) +func (s *Service) AutoAllowed(svc model.Service) error +func FollowedRef(actualRef string) string // "repo:tag", or "" when no tag is followed +``` + +`Start` has the return contract `Trigger` had: `*model.NotUpdatableError` with no job, `model.ErrJobActive` with the existing job, and an already-failed job for offline / too-old / start failures. `Trigger(ctx, id, by)` is `GetService` plus `Start` with `TriggerManual`. + +### C5. `internal/master/webhook` + +```go +type Config struct { Token string; Timeout time.Duration; PublicURL string } +func New(store Store, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) *Handler +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) +``` + +Request `{"image","tag"}`; response `{"request_id","image","tag","matched","results":[{"service_id","service","kind","mode","status","job_id","job_url","to_ref","error","duration_ms"}]}`. Statuses `succeeded | failed | timeout | marked | busy | refused`; HTTP `500` when any result is `failed` or `timeout`, else `200`. Other codes: `400`, `401`, `404`, `503`. See the README for the full table. + +### C6. Master configuration + +`UPDATER_API_KEY` (optional) and `UPDATE_TIMEOUT` (default `5m`, positive Go duration, same error text style as the agent's). + +### C7. Web + +`POST /services/{id}/mode` with `mode=manual|auto` and optional `return=overview`. `303` to the service page (or `/`), `400` for a bad mode, `404` for an unknown or removed service, `409` re-rendering the service page with "Cannot enable auto: ." Publishes `events.Services`. + +## Tasks + +### Task 1: Store and model + +Migration C1, model C2, store C3. Integration tests (`//go:build integration`): `ServicesByRepo` excludes removed rows and other repos; `SetAvailable` round trip; `SetUpdateMode` including `ErrNotFound`; new service is manual and upsert/snapshot keep the mode; `FinishJob` clears the same ref, keeps a different ref, keeps newer, keeps on failure. + +### Task 2: Jobs service + +Extract `Start` from `Trigger`; `updatable(svc, explicitRef)`; `AutoAllowed`; `FollowedRef`. Tests: webhook trigger with explicit ref is recorded; explicit ref bypasses only the no-tag refusal; `AutoAllowed` table; existing `TestTrigger*` stay green. + +### Task 3: Config, webhook, wiring + +Config C6 with tests. Handler C5 with fakes and the real `events.Broker`. `cmd/master/main.go` mounts `POST /update` directly on the mux and logs `webhook_disabled` when the key is empty. + +### Task 4: Web + +Store gains `SetUpdateMode`, Jobs gains `AutoAllowed`. Overview: mode column with a toggle form (hidden `csrf_token`, `mode`, `return=overview`; disabled with the reason as visible text when auto is not allowed) and an "update available" badge. Service page: Available row, Mode row, toggle form, 409 error. The forms are inside the `services` template that the SSE fragment re-renders, so they survive a refresh. + +### Task 5: End-to-end + +`internal/e2e/webhook_test.go`: `TestWebhookWaitsForAutoJob` (real in-process master and agent, gate executor, the handler on an `httptest` server) and `TestWebhookFailedAutoJobIs500`. The in-memory `memStore` gains `ServicesByRepo`, `SetAvailable` and the same clearing rule as the SQL. + +### Task 6: Documentation + +README (webhook section, master variables, warning about `auto`), `gitea-action/README.md` (do not change `action.yml`), `deploy/master.stack.yml` (`UPDATER_API_KEY` as a Docker secret), `CLAUDE.md`, this plan. + +## Final verification + +```bash +gofmt -l . ; go vet ./... +go build ./cmd/master ./cmd/agent && rm -f master agent +go test ./... +go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/ +DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ # when a Postgres is available +``` + +Manual smoke test: set `UPDATER_API_KEY`, make a throwaway Compose service `auto`, then `curl -X POST -H "Authorization: Bearer $KEY" -d '{"image":"","tag":""}' https:///update`. Expect `200` with `succeeded` after the job finishes, the job visible in the UI with trigger `webhook`, and the "update available" badge gone. diff --git a/gitea-action/README.md b/gitea-action/README.md index bc0816a..ef8475c 100644 --- a/gitea-action/README.md +++ b/gitea-action/README.md @@ -1,6 +1,6 @@ # Deploy via package-updater (composite action) -Notifies `package-updater` to `docker compose pull` + `up -d` for the matching service(s) after a CI build. +Notifies the `package-updater` master after a CI build. The master records the pushed version on every matching service and, for services in auto mode, updates them and waits for the result. ## Usage @@ -20,7 +20,7 @@ jobs: token: ${{ secrets.UPDATER_TOKEN }} ``` -`UPDATER_TOKEN` should be set as an organisation-level secret in Gitea so all repos share it. +`endpoint` is the master's `/update`, and `UPDATER_TOKEN` is the master's `UPDATER_API_KEY`. It should be set as an organisation-level secret in Gitea so all repos share it. ## Inputs @@ -28,9 +28,20 @@ jobs: |---|---|---|---| | `endpoint` | yes | — | Full URL to `/update` | | `image` | yes | — | Image reference without tag | -| `tag` | no | `""` | Tag that was just pushed (logged for audit) | +| `tag` | no | `""` | Tag that was just pushed. Auto services are updated to `:`, so it must exist in the registry. Omit it to re-pull the tag each service already follows. | | `token` | yes | — | Bearer token configured in package-updater | ## Failure modes -The step exits non-zero if `package-updater` returns HTTP 4xx or 5xx. This is intentional — the workflow surfaces the deploy failure to whoever pushed. +The step exits non-zero on any HTTP 4xx or 5xx. This is intentional: the workflow surfaces the deploy failure to whoever pushed. + +| Code | Meaning | +|---|---| +| `200` | Every match was marked, updated, busy or refused. Manual services are only marked, which is not an error. A service that follows no tag is `refused` when the request has no `tag`. | +| `401` | Wrong or missing token. | +| `404` | No service uses the image. Check `image` for typos; matching is case-sensitive. | +| `500` | At least one auto job failed or timed out (`UPDATE_TIMEOUT` on the master). | +| `503` | The master has no `UPDATER_API_KEY`. | + +A request that matches both a working and a failed service now fails as a whole (`500`); the old single-instance updater's `207` no longer exists. +The master waits for auto jobs, so the request can take minutes; raise any step timeout in your workflow accordingly. diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index 905b064..9036eb1 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -11,6 +11,7 @@ import ( "net" "net/http" "net/http/httptest" + "sort" "strings" "sync" "sync/atomic" @@ -65,6 +66,33 @@ func (m *memStore) GetService(_ context.Context, id int64) (model.Service, error return svc, nil } +// ServicesByRepo returns the seeded services whose image repo equals repo, +// ordered by ID. +func (m *memStore) ServicesByRepo(_ context.Context, repo string) ([]model.Service, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []model.Service + for _, svc := range m.svcRows { + if svc.ImageRepo == repo && svc.RemovedAt == nil { + out = append(out, svc) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +func (m *memStore) SetAvailable(_ context.Context, id int64, a model.Available) error { + m.mu.Lock() + defer m.mu.Unlock() + svc, ok := m.svcRows[id] + if !ok { + return model.ErrNotFound + } + svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Source, &a.At + m.svcRows[id] = svc + return nil +} + func (m *memStore) CreateJob(_ context.Context, nj model.NewJob) (model.Job, error) { m.mu.Lock() defer m.mu.Unlock() @@ -146,6 +174,7 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo return false, nil } j.Status, j.Error, j.Output, j.OutputTruncated, j.FinishedAt = f.Status, f.Error, f.Output, f.OutputTruncated, &f.FinishedAt + requestedRef := j.ToRef if f.Ref != "" { j.ToRef = f.Ref } @@ -155,6 +184,11 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo if f.Status == model.JobSucceeded { svc := m.svcRows[j.ServiceID] svc.DesiredRef = f.Ref + // Same rule as the SQL: the job was created for the suggested ref after it was reported. + requested, _, _ := strings.Cut(requestedRef, "@") + if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) { + svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = "", "", nil + } m.svcRows[j.ServiceID] = svc } m.finishApplied++ diff --git a/internal/e2e/jobs_test.go b/internal/e2e/jobs_test.go index d3f1ae7..912ea93 100644 --- a/internal/e2e/jobs_test.go +++ b/internal/e2e/jobs_test.go @@ -102,7 +102,7 @@ func newJobEnv(t *testing.T) *jobEnv { t.Helper() store := &memStore{agents: map[string]model.Agent{agenttoken.Hash("pua_ok"): {ID: agentID, Name: "node-2"}}} store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", - AgentID: ptr(agentID), ActualRef: slusk + ":1.4.2@sha256:old"}) + AgentID: ptr(agentID), ImageRepo: slusk, ActualRef: slusk + ":1.4.2@sha256:old"}) m := startMaster(t, store) exec := newGateExecutor() env := &jobEnv{store: store, m: m, exec: exec, diff --git a/internal/e2e/webhook_test.go b/internal/e2e/webhook_test.go new file mode 100644 index 0000000..5e38c82 --- /dev/null +++ b/internal/e2e/webhook_test.go @@ -0,0 +1,128 @@ +package e2e + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/master/webhook" +) + +// webhookServer mounts the webhook handler on a test server, wired to the +// job environment's store, jobs service and broker. +func (e *jobEnv) webhookServer(t *testing.T) *httptest.Server { + t.Helper() + h := webhook.New(e.store, e.jobs, e.m.broker, webhook.Config{ + Token: "ci-key", Timeout: 5 * time.Second, PublicURL: "https://updater.example.com", + }, slog.New(slog.NewTextHandler(io.Discard, nil))) + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + return srv +} + +func (e *jobEnv) setAuto(id int64) { + e.store.mu.Lock() + defer e.store.mu.Unlock() + svc := e.store.svcRows[id] + svc.UpdateMode = model.ModeAuto + e.store.svcRows[id] = svc +} + +type webhookResponse struct { + Matched int `json:"matched"` + Results []struct { + Status string `json:"status"` + JobID int64 `json:"job_id"` + ToRef string `json:"to_ref"` + Error string `json:"error"` + } `json:"results"` +} + +func callWebhook(t *testing.T, srv *httptest.Server, body string) (int, webhookResponse) { + t.Helper() + req, err := http.NewRequest(http.MethodPost, srv.URL, strings.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer ci-key") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + var out webhookResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&out)) + return resp.StatusCode, out +} + +func TestWebhookWaitsForAutoJob(t *testing.T) { + e := newJobEnv(t) + e.setAuto(1) + srv := e.webhookServer(t) + + type reply struct { + code int + body webhookResponse + } + done := make(chan reply, 1) + go func() { + code, body := callWebhook(t, srv, `{"image":"`+slusk+`:ignored","tag":"2.0.0"}`) + done <- reply{code, body} + }() + + e.waitStarted(t) + select { + case <-done: + t.Fatal("the webhook answered before the job finished") + case <-time.After(100 * time.Millisecond): + } + close(e.exec.release) + + var r reply + select { + case r = <-done: + case <-time.After(waitFor): + t.Fatal("the webhook never answered") + } + require.Equal(t, http.StatusOK, r.code) + require.Len(t, r.body.Results, 1) + assert.Equal(t, "succeeded", r.body.Results[0].Status) + assert.Equal(t, slusk+":2.0.0", r.body.Results[0].ToRef) + + job := e.job(t, r.body.Results[0].JobID) + assert.Equal(t, model.TriggerWebhook, job.Trigger) + assert.Equal(t, model.JobSucceeded, job.Status) + assert.Equal(t, slusk+":2.0.0@sha256:e2e", job.ToRef) + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Empty(t, svc.AvailableRef, "the job deployed the suggestion, so it is cleared") +} + +func TestWebhookFailedAutoJobIs500(t *testing.T) { + e := newJobEnv(t) + e.setAuto(1) + srv := e.webhookServer(t) + e.agent.stopConn() // no agent connection: the job fails at once with "agent offline" + require.Eventually(t, func() bool { + svc, _ := e.store.GetService(context.Background(), 1) + _, err := e.m.hub.Route(svc) + return err != nil + }, waitFor, pollEvery) + + code, body := callWebhook(t, srv, `{"image":"`+slusk+`","tag":"2.0.0"}`) + + assert.Equal(t, http.StatusInternalServerError, code) + require.Len(t, body.Results, 1) + assert.Equal(t, "failed", body.Results[0].Status) + assert.Equal(t, "agent offline", body.Results[0].Error) + assert.Equal(t, model.JobFailed, e.job(t, body.Results[0].JobID).Status) + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Equal(t, slusk+":2.0.0", svc.AvailableRef, "a failed job keeps the suggestion") +} diff --git a/internal/master/config/config.go b/internal/master/config/config.go index 0739ec1..77acd6f 100644 --- a/internal/master/config/config.go +++ b/internal/master/config/config.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strings" + "time" ) // Config holds the master's runtime configuration. @@ -20,8 +21,18 @@ type Config struct { Port string AgentPort string LogLevel string + + // UpdaterAPIKey is the bearer token for POST /update (UPDATER_API_KEY). + // Empty disables the webhook. + UpdaterAPIKey string + // UpdateTimeout bounds how long one webhook request waits for its auto + // jobs (UPDATE_TIMEOUT, default 5m). + UpdateTimeout time.Duration } +// defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset. +const defaultUpdateTimeout = 5 * time.Minute + // Load reads and validates the master configuration, failing fast on // anything missing so a half-configured master never starts. func Load() (*Config, error) { @@ -37,6 +48,8 @@ func Load() (*Config, error) { Port: getenvDefault("PORT", "8080"), AgentPort: getenvDefault("AGENT_PORT", "8090"), LogLevel: getenvDefault("LOG_LEVEL", "info"), + UpdaterAPIKey: os.Getenv("UPDATER_API_KEY"), + UpdateTimeout: defaultUpdateTimeout, } for _, r := range []struct{ name, value string }{ {"DATABASE_URL", cfg.DatabaseURL}, @@ -57,6 +70,13 @@ func Load() (*Config, error) { if !strings.HasPrefix(cfg.AgentURL, "wss://") { return nil, fmt.Errorf("AGENT_URL must start with wss://") } + 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 + } return cfg, nil } diff --git a/internal/master/config/config_test.go b/internal/master/config/config_test.go index 3ef5614..977ee67 100644 --- a/internal/master/config/config_test.go +++ b/internal/master/config/config_test.go @@ -2,6 +2,7 @@ package config import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -21,7 +22,7 @@ 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"} { + for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT"} { t.Setenv(k, "") } } @@ -35,6 +36,29 @@ func TestLoad_Defaults(t *testing.T) { 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) + assert.Empty(t, cfg.UpdaterAPIKey) + assert.Equal(t, 5*time.Minute, cfg.UpdateTimeout) +} + +func TestLoad_WebhookSettings(t *testing.T) { + setAll(t) + t.Setenv("UPDATER_API_KEY", "s3cret") + t.Setenv("UPDATE_TIMEOUT", "90s") + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, "s3cret", cfg.UpdaterAPIKey) + assert.Equal(t, 90*time.Second, cfg.UpdateTimeout) +} + +func TestLoad_InvalidUpdateTimeout(t *testing.T) { + for _, v := range []string{"abc", "0s", "-5m", "30"} { + t.Run(v, func(t *testing.T) { + setAll(t) + t.Setenv("UPDATE_TIMEOUT", v) + _, err := Load() + assert.ErrorContains(t, err, "UPDATE_TIMEOUT must be a positive duration") + }) + } } func TestLoad_EachRequiredVariable(t *testing.T) { diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go index 49086c7..6b2fe1b 100644 --- a/internal/master/jobs/jobs.go +++ b/internal/master/jobs/jobs.go @@ -68,9 +68,34 @@ func target(actualRef string) (repo, tag string, ok bool) { return ref.Repo, "", false } +// FollowedRef returns "repo:tag" for the tag a service with this actual ref +// follows (a bare image follows "latest"), or "" when it follows no tag. +func FollowedRef(actualRef string) string { + repo, tag, ok := target(actualRef) + if !ok { + return "" + } + return repo + ":" + tag +} + // Updatable returns nil or a *model.NotUpdatableError. It looks only at the -// service row, so the UI can call it for every page render. +// service row, so the UI can call it for every page render. It answers for an +// update to the service's own followed tag (the Update button). func (s *Service) Updatable(svc model.Service) error { + return s.updatable(svc, false) +} + +// AutoAllowed returns nil or a *model.NotUpdatableError saying why the +// service can never be updated automatically. Unlike Updatable it does not +// refuse a service that follows no tag: a webhook that names a tag can still +// update it. +func (s *Service) AutoAllowed(svc model.Service) error { + return s.updatable(svc, true) +} + +// updatable checks the service row. explicitRef means the caller supplies the +// target ref, so a service that follows no tag is fine. +func (s *Service) updatable(svc model.Service, explicitRef bool) error { switch { case svc.RemovedAt != nil: return &model.NotUpdatableError{Reason: model.ReasonRemoved} @@ -81,32 +106,49 @@ func (s *Service) Updatable(svc model.Service) error { case svc.Kind == protocol.KindContainer: return &model.NotUpdatableError{Reason: model.ReasonContainer} } - if _, _, ok := target(svc.ActualRef); !ok { + if _, _, ok := target(svc.ActualRef); !ok && !explicitRef { return &model.NotUpdatableError{Reason: model.ReasonNoTag} } return nil } +// StartRequest says who asks for an update and what to deploy. +type StartRequest struct { + Trigger model.JobTrigger + RequestedBy string // OIDC subject for manual jobs; "" otherwise + // Ref is the ref to deploy ("repo:tag"). Empty means the service's + // followed tag, derived from its actual ref. + Ref string +} + // Trigger starts a manual update of the service to its current tag. It -// returns the job, which may already be failed ("agent offline", "agent too -// old for updates"); a *model.NotUpdatableError when the service cannot be -// updated (no job created); the active job and model.ErrJobActive when one -// is queued or running; model.ErrNotFound for an unknown service. +// returns what Start returns, and model.ErrNotFound for an unknown service. func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) { svc, err := s.store.GetService(ctx, serviceID) if err != nil { return model.Job{}, err } - if err := s.Updatable(svc); err != nil { + return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy}) +} + +// Start starts an update of svc. It returns the job, which may already be +// failed ("agent offline", "agent too old for updates"); a +// *model.NotUpdatableError when the service cannot be updated (no job +// created); the active job and model.ErrJobActive when one is queued or +// running. +func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) { + if err := s.updatable(svc, req.Ref != ""); err != nil { return model.Job{}, err } - repo, tag, _ := target(svc.ActualRef) // Updatable checked the tag - ref := repo + ":" + tag + ref := req.Ref + if ref == "" { + ref = FollowedRef(svc.ActualRef) // updatable checked the tag + } r, routeErr := s.hub.Route(svc) nj := model.NewJob{ ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref, - Trigger: model.TriggerManual, RequestedBy: requestedBy, CreatedAt: s.now(), + Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(), } if routeErr == nil { nj.AgentID = &r.AgentID diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go index 5e7468a..3918fb2 100644 --- a/internal/master/jobs/jobs_test.go +++ b/internal/master/jobs/jobs_test.go @@ -242,3 +242,102 @@ func TestTrigger_NotUpdatableCreatesNoJob(t *testing.T) { assert.Empty(t, h.starts) assert.Empty(t, pub.topics) } + +func TestStart_WebhookTriggerWithExplicitRef(t *testing.T) { + st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + job, err := s.Start(context.Background(), composeSvc, StartRequest{ + Trigger: model.TriggerWebhook, Ref: "gitea.shcizo.se/shcizo/slusk:2.0.0"}) + + require.NoError(t, err) + assert.Equal(t, model.JobQueued, job.Status) + require.Len(t, st.created, 1) + assert.Equal(t, model.TriggerWebhook, st.created[0].Trigger) + assert.Empty(t, st.created[0].RequestedBy) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", st.created[0].ToRef) + require.Len(t, h.starts, 1) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", h.starts[0].start.Ref) +} + +func TestStart_ExplicitRefBypassesOnlyTheNoTagRefusal(t *testing.T) { + tests := []struct { + name string + svc model.Service + reason string // "" means the job starts + }{ + {"digest only", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""}, + {"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved}, + {"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf}, + {"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt}, + {"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tt.svc.ID = 1 + st := &fakeStore{services: map[int64]model.Service{1: tt.svc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + _, err := s.Start(context.Background(), tt.svc, StartRequest{Trigger: model.TriggerWebhook, Ref: "x/y:2"}) + + if tt.reason == "" { + require.NoError(t, err) + assert.Len(t, st.created, 1) + return + } + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Equal(t, tt.reason, nu.Reason) + assert.Empty(t, st.created) + }) + } +} + +func TestStart_WithoutRefStillRefusesNoTag(t *testing.T) { + svc := model.Service{ID: 1, Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"} + s, _ := newService(&fakeStore{jobs: map[int64]model.Job{}}, &fakeHub{}) + + _, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerWebhook}) + + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Equal(t, model.ReasonNoTag, nu.Reason) +} + +func TestAutoAllowed(t *testing.T) { + tests := []struct { + name string + svc model.Service + reason string + }{ + {"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved}, + {"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf}, + {"agent image repo", model.Service{Kind: protocol.KindCompose, ActualRef: selfImage}, model.ReasonSelf}, + {"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt}, + {"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer}, + {"digest without tag is allowed", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""}, + {"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""}, + } + s, _ := newService(&fakeStore{}, &fakeHub{}) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := s.AutoAllowed(tt.svc) + if tt.reason == "" { + assert.NoError(t, err) + return + } + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Equal(t, tt.reason, nu.Reason) + }) + } +} + +func TestFollowedRef(t *testing.T) { + assert.Equal(t, "postgres:16", FollowedRef("postgres:16@sha256:abc")) + assert.Equal(t, "postgres:latest", FollowedRef("postgres")) + assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo")) + assert.Empty(t, FollowedRef("redis@sha256:abc")) +} diff --git a/internal/master/model/job.go b/internal/master/model/job.go index afd2106..5c1f8b9 100644 --- a/internal/master/model/job.go +++ b/internal/master/model/job.go @@ -46,7 +46,7 @@ const ( JobFailed JobStatus = "failed" ) -// JobTrigger says what created a job. Only TriggerManual is used in phase 2. +// JobTrigger says what created a job. TriggerManual comes from the UI, TriggerWebhook from POST /update. type JobTrigger string // Job triggers. diff --git a/internal/master/model/model.go b/internal/master/model/model.go index 086402d..3b4a8ff 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -51,11 +51,34 @@ type Service struct { ActualDigest string Status string Self bool - Built bool // Compose service with a build: section; never updated - UpdateMode string - FirstSeenAt time.Time - UpdatedAt time.Time - RemovedAt *time.Time + Built bool // Compose service with a build: section; never updated + UpdateMode string // ModeManual or ModeAuto + // AvailableRef is the newest ref a webhook or poll reported for this + // service; "" if none, or if a later job already deployed it. + AvailableRef string + AvailableSource string // SourceWebhook; "" when AvailableRef is empty + AvailableAt *time.Time + FirstSeenAt time.Time + UpdatedAt time.Time + RemovedAt *time.Time +} + +// Update modes of a service. +const ( + ModeManual = "manual" + ModeAuto = "auto" +) + +// Sources of an available ref. +const ( + SourceWebhook = "webhook" +) + +// Available is a newer ref reported for a service. +type Available struct { + Ref string + Source string // SourceWebhook + At time.Time } // DisplayName is the human-facing name of the service. diff --git a/internal/master/store/jobs.go b/internal/master/store/jobs.go index a118c43..31a9da9 100644 --- a/internal/master/store/jobs.go +++ b/internal/master/store/jobs.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "time" "github.com/jackc/pgx/v5" @@ -187,7 +188,10 @@ func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk str // agentID: status, error, output (replacing what progress assembled), // started_at (the agent's, or the existing one if the agent sent none), // finished_at; to_ref = f.Ref when set. On success it also sets the -// service's desired_ref = f.Ref, in the same transaction. Returns false if no +// service's desired_ref = f.Ref and clears its available ref when that is +// the ref the job was created for (ignoring any @digest) and was reported no +// later than the job was created (that version is now running), in the same +// transaction. Returns false if no // such active job belongs to agentID. func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) { tx, err := s.pool.Begin(ctx) @@ -203,14 +207,18 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job startedAt = &f.StartedAt } var serviceID int64 + var createdAt time.Time + var requestedRef string + // requested is the job's to_ref as it was created, before f.Ref overwrites it. err = tx.QueryRow(ctx, ` - UPDATE update_jobs SET + UPDATE update_jobs j SET status = $3, error = $4, output = $5, output_truncated = $6, - to_ref = CASE WHEN $7::text <> '' THEN $7::text ELSE to_ref END, - started_at = COALESCE($8, started_at), finished_at = $9 - WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running') - RETURNING service_id`, - jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, startedAt, f.FinishedAt).Scan(&serviceID) + to_ref = CASE WHEN $7::text <> '' THEN $7::text ELSE j.to_ref END, + started_at = COALESCE($8, j.started_at), finished_at = $9 + FROM (SELECT id, to_ref FROM update_jobs WHERE id = $1) requested + WHERE j.id = requested.id AND j.agent_id = $2 AND j.status IN ('queued', 'running') + RETURNING j.service_id, j.created_at, requested.to_ref`, + jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, startedAt, f.FinishedAt).Scan(&serviceID, &createdAt, &requestedRef) if errors.Is(err, pgx.ErrNoRows) { return false, nil } @@ -218,7 +226,13 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job return false, fmt.Errorf("finish job: %w", err) } if f.Status == model.JobSucceeded { - if _, err := tx.Exec(ctx, `UPDATE services SET desired_ref = $2 WHERE id = $1`, serviceID, f.Ref); err != nil { + requestedRef, _, _ = strings.Cut(requestedRef, "@") + if _, err := tx.Exec(ctx, ` + UPDATE services SET desired_ref = $2, + available_ref = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_ref END, + available_source = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_source END, + available_at = CASE WHEN available_ref = $4 AND available_at <= $3 THEN NULL ELSE available_at END + WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef); err != nil { return false, fmt.Errorf("finish job: desired ref: %w", err) } } diff --git a/internal/master/store/migrations/0003_webhook.sql b/internal/master/store/migrations/0003_webhook.sql new file mode 100644 index 0000000..0ea2bc2 --- /dev/null +++ b/internal/master/store/migrations/0003_webhook.sql @@ -0,0 +1,6 @@ +ALTER TABLE services ADD COLUMN available_ref text NOT NULL DEFAULT ''; +ALTER TABLE services ADD COLUMN available_source text NOT NULL DEFAULT '' + CHECK (available_source IN ('', 'webhook')); +ALTER TABLE services ADD COLUMN available_at timestamptz; + +CREATE INDEX services_by_image_repo ON services (image_repo) WHERE removed_at IS NULL; diff --git a/internal/master/store/services.go b/internal/master/store/services.go index 816d2c5..a71f6d5 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -149,7 +149,8 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), s.compose_project, s.compose_service, s.working_dir, s.config_files, s.built, s.swarm_cluster_id, s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest, - s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at, s.removed_at` + s.status, s.self, s.update_mode, s.available_ref, s.available_source, s.available_at, + s.first_seen_at, s.updated_at, s.removed_at` func scanService(row pgx.Row) (model.Service, error) { var m model.Service @@ -157,7 +158,8 @@ func scanService(row pgx.Row) (model.Service, error) { err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, &m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.Built, &m.SwarmClusterID, &m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest, - &m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + &m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableSource, &m.AvailableAt, + &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) m.Kind = protocol.Kind(kind) return m, err } @@ -165,13 +167,23 @@ func scanService(row pgx.Row) (model.Service, error) { // ListServices returns all services not marked removed: machine-bound // services grouped by agent name first, Swarm services last. func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) { + return s.queryServices(ctx, "list services", "", nil) +} + +// ServicesByRepo returns the services not marked removed whose image repo +// equals repo (case-sensitive), in the same order as ListServices. +func (s *Store) ServicesByRepo(ctx context.Context, repo string) ([]model.Service, error) { + return s.queryServices(ctx, "services by repo", "AND s.image_repo = $1", []any{repo}) +} + +func (s *Store) queryServices(ctx context.Context, what, extraWhere string, args []any) ([]model.Service, error) { rows, err := s.pool.Query(ctx, ` SELECT `+serviceColumns+` FROM services s LEFT JOIN agents a ON a.id = s.agent_id - WHERE s.removed_at IS NULL - ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`) + WHERE s.removed_at IS NULL `+extraWhere+` + ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`, args...) if err != nil { - return nil, fmt.Errorf("list services: %w", err) + return nil, fmt.Errorf("%s: %w", what, err) } defer rows.Close() var out []model.Service @@ -185,6 +197,32 @@ func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) { return out, rows.Err() } +// SetAvailable records a newer ref for the service, replacing any earlier +// suggestion. +func (s *Store) SetAvailable(ctx context.Context, id int64, a model.Available) error { + if _, err := s.pool.Exec(ctx, ` + UPDATE services SET available_ref = $2, available_source = $3, available_at = $4 + WHERE id = $1`, id, a.Ref, a.Source, a.At); err != nil { + return fmt.Errorf("set available: %w", err) + } + return nil +} + +// SetUpdateMode sets the service's update mode (model.ModeManual or +// model.ModeAuto). Returns model.ErrNotFound if the service does not exist +// or is marked removed. +func (s *Store) SetUpdateMode(ctx context.Context, id int64, mode string) error { + tag, err := s.pool.Exec(ctx, + `UPDATE services SET update_mode = $2 WHERE id = $1 AND removed_at IS NULL`, id, mode) + if err != nil { + return fmt.Errorf("set update mode: %w", err) + } + if tag.RowsAffected() == 0 { + return model.ErrNotFound + } + return nil +} + // GetService returns one service, including a removed one (RemovedAt set). // Returns model.ErrNotFound if the ID does not exist. func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) { diff --git a/internal/master/store/webhook_integration_test.go b/internal/master/store/webhook_integration_test.go new file mode 100644 index 0000000..e047cec --- /dev/null +++ b/internal/master/store/webhook_integration_test.go @@ -0,0 +1,153 @@ +//go:build integration + +package store + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +func TestServicesByRepo(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "o", ComposeService: "o", Image: "gitea.shcizo.se/shcizo/Slusk:1", Status: "running"} + gone := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "g", ComposeService: "g", Image: "gitea.shcizo.se/shcizo/slusk:0.9", Status: "running"} + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, other, gone, adhoc}, t0)) + require.NoError(t, s.RemoveService(ctx, id, "", gone.Key(), t0)) + + got, err := s.ServicesByRepo(ctx, "gitea.shcizo.se/shcizo/slusk") + + require.NoError(t, err) + assert.Equal(t, []string{"slusk/web"}, names(got), "removed rows and rows of other (case-sensitive) repos are excluded") + assert.Equal(t, "node-2", got[0].AgentName) + + none, err := s.ServicesByRepo(ctx, "nope") + require.NoError(t, err) + assert.Empty(t, none) +} + +func TestSetAvailable_RoundTrip(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + _, svc := jobFixture(t, s, "a") + + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "r:2", Source: model.SourceWebhook, At: t0})) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "r:2", got.AvailableRef) + assert.Equal(t, model.SourceWebhook, got.AvailableSource) + require.NotNil(t, got.AvailableAt) + assert.True(t, t0.Equal(*got.AvailableAt)) +} + +func TestSetUpdateMode(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto)) + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, model.ModeAuto, got.UpdateMode) + + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeManual)) + got, _ = s.GetService(ctx, svc) + assert.Equal(t, model.ModeManual, got.UpdateMode) + + assert.ErrorIs(t, s.SetUpdateMode(ctx, 99999, model.ModeAuto), model.ErrNotFound) + + require.NoError(t, s.RemoveService(ctx, agent, "", slusk.Key(), t0)) + assert.ErrorIs(t, s.SetUpdateMode(ctx, svc, model.ModeAuto), model.ErrNotFound) +} + +func TestUpsertKeepsUpdateMode(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, model.ModeManual, got.UpdateMode, "a new service starts manual") + + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto)) + changed := slusk + changed.Image = "gitea.shcizo.se/shcizo/slusk:1.5.0" + require.NoError(t, s.UpsertService(ctx, agent, "", changed, t0.Add(time.Minute))) + require.NoError(t, s.ApplySnapshot(ctx, agent, worker, []protocol.Service{changed}, t0.Add(2*time.Minute))) + + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, model.ModeAuto, got.UpdateMode) +} + +func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) { + succeed := model.JobFinish{Status: model.JobSucceeded, Ref: "r:2@sha256:d", StartedAt: t0, FinishedAt: t0} + tests := []struct { + name string + availableRef string + availableAt time.Time // job is created at t0 for r:2 + finish model.JobFinish + wantCleared bool + }{ + {"same ref, suggestion before the job", "r:2", t0.Add(-time.Minute), succeed, true}, + {"same ref, suggestion at the same instant", "r:2", t0, succeed, true}, + {"same ref, suggestion after the job was created", "r:2", t0.Add(time.Minute), succeed, false}, + {"different ref keeps the suggestion", "r:3", t0.Add(-time.Minute), succeed, false}, + {"failed job keeps the suggestion", "r:2", t0.Add(-time.Minute), + model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Source: model.SourceWebhook, At: tt.availableAt})) + nj := newJob(svc, agent) + nj.ToRef = "r:2" + j, err := s.CreateJob(ctx, nj) + require.NoError(t, err) + + ok, err := s.FinishJob(ctx, agent, j.ID, tt.finish) + require.NoError(t, err) + require.True(t, ok) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + if tt.wantCleared { + assert.Empty(t, got.AvailableRef) + assert.Empty(t, got.AvailableSource) + assert.Nil(t, got.AvailableAt) + } else { + assert.Equal(t, tt.availableRef, got.AvailableRef) + assert.Equal(t, model.SourceWebhook, got.AvailableSource) + assert.NotNil(t, got.AvailableAt) + } + }) + } +} + +func TestFinishJob_WithoutAvailableLeavesItEmpty(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + j, err := s.CreateJob(ctx, newJob(svc, agent)) + require.NoError(t, err) + + ok, err := s.FinishJob(ctx, agent, j.ID, model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", StartedAt: t0, FinishedAt: t0}) + require.NoError(t, err) + require.True(t, ok) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Empty(t, got.AvailableRef) + assert.Nil(t, got.AvailableAt) + assert.Equal(t, "r:1@sha256:d", got.DesiredRef) +} diff --git a/internal/master/web/server.go b/internal/master/web/server.go index c1cef5b..58434c1 100644 --- a/internal/master/web/server.go +++ b/internal/master/web/server.go @@ -32,6 +32,8 @@ var agentNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`) type Store interface { ListServices(ctx context.Context) ([]model.Service, error) GetService(ctx context.Context, id int64) (model.Service, error) + // SetUpdateMode returns model.ErrNotFound for an unknown or removed service. + SetUpdateMode(ctx context.Context, id int64, mode string) 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) @@ -46,10 +48,11 @@ type Hub interface { Disconnect(agentID int64) } -// Jobs starts updates and explains why a service cannot be updated. -// *jobs.Service satisfies it. +// Jobs starts updates and explains why a service cannot be updated or set +// to auto mode. *jobs.Service satisfies it. type Jobs interface { Updatable(svc model.Service) error + AutoAllowed(svc model.Service) error Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) } @@ -104,6 +107,7 @@ func (s *Server) Register(mux *http.ServeMux, protect func(http.Handler) http.Ha route("GET /services/{id}", s.servicePage) route("GET /fragments/services/{id}/jobs", s.serviceJobsFragment) route("POST /services/{id}/update", s.updateService) + route("POST /services/{id}/mode", s.setMode) route("GET /jobs", s.jobsPage) route("GET /fragments/jobs", s.jobsFragment) route("GET /jobs/{id}", s.jobPage) @@ -127,6 +131,7 @@ type pageData struct { Service model.Service ConfigFiles []string Reason string // why Service cannot be updated; "" when it can + AutoBlocked string // why Service cannot be set to auto; "" when it can Jobs []model.Job Job model.Job } @@ -150,7 +155,7 @@ func (s *Server) serviceData(r *http.Request) (pageData, error) { if err != nil { return data, err } - data.Groups = groupServices(services) + data.Groups = groupServices(services, s.autoBlocked) return data, nil } @@ -172,6 +177,15 @@ func (s *Server) servicesFragment(w http.ResponseWriter, r *http.Request) { s.render(w, http.StatusOK, "overview.html", "services", data) } +// autoBlocked returns the reason svc cannot be set to auto mode, or "". +func (s *Server) autoBlocked(svc model.Service) string { + var nu *model.NotUpdatableError + if err := s.jobs.AutoAllowed(svc); errors.As(err, &nu) { + return nu.Reason + } + return "" +} + // 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) @@ -207,6 +221,7 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) { } else if err != nil { return data, err } + data.AutoBlocked = s.autoBlocked(svc) data.Jobs, err = s.store.ListJobs(r.Context(), model.JobFilter{ServiceID: id, Limit: serviceJobsLimit}) return data, err } @@ -263,6 +278,54 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) { } } +// setMode switches a service between manual and auto. Manual is always +// allowed; auto only when jobs.AutoAllowed says so. +func (s *Server) setMode(w http.ResponseWriter, r *http.Request) { + id, ok := pathID(w, r) + if !ok { + return + } + mode := r.PostFormValue("mode") + if mode != model.ModeManual && mode != model.ModeAuto { + http.Error(w, "invalid mode", http.StatusBadRequest) + return + } + if mode == model.ModeAuto { + svc, err := s.store.GetService(r.Context(), id) + if err == nil && svc.RemovedAt != nil { + err = model.ErrNotFound + } + if err != nil { + s.failLookup(w, err) + return + } + var nu *model.NotUpdatableError + if err := s.jobs.AutoAllowed(svc); errors.As(err, &nu) { + data, derr := s.serviceDetail(r, id) + if derr != nil { + s.failLookup(w, derr) + return + } + data.Error = "Cannot enable auto: " + nu.Reason + "." + s.render(w, http.StatusConflict, "service.html", "layout", data) + return + } else if err != nil { + s.fail(w, err) + return + } + } + if err := s.store.SetUpdateMode(r.Context(), id, mode); err != nil { + s.failLookup(w, err) + return + } + s.broker.Publish(events.Services) + if r.PostFormValue("return") == "overview" { + http.Redirect(w, r, "/", http.StatusSeeOther) + return + } + http.Redirect(w, r, "/services/"+strconv.FormatInt(id, 10), http.StatusSeeOther) +} + func (s *Server) jobsData(r *http.Request) (pageData, error) { data := s.base(r, "Jobs") var err error diff --git a/internal/master/web/server_test.go b/internal/master/web/server_test.go index 43b11e8..52e40b4 100644 --- a/internal/master/web/server_test.go +++ b/internal/master/web/server_test.go @@ -46,6 +46,29 @@ func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, erro return model.Service{}, model.ErrNotFound } +func (f *fakeStore) SetUpdateMode(_ context.Context, id int64, mode string) error { + f.mu.Lock() + defer f.mu.Unlock() + for i := range f.services { + if f.services[i].ID == id && f.services[i].RemovedAt == nil { + f.services[i].UpdateMode = mode + return nil + } + } + return model.ErrNotFound +} + +func (f *fakeStore) modeOf(id int64) string { + f.mu.Lock() + defer f.mu.Unlock() + for _, s := range f.services { + if s.ID == id { + return s.UpdateMode + } + } + return "" +} + func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { for _, j := range f.jobs { if j.ID == id { @@ -99,6 +122,7 @@ func (f *fakeStore) revoked() int64 { // fakeJobs returns scripted results and records who asked for the update. type fakeJobs struct { updatable error + autoErr error job model.Job err error @@ -108,6 +132,8 @@ type fakeJobs struct { func (f *fakeJobs) Updatable(model.Service) error { return f.updatable } +func (f *fakeJobs) AutoAllowed(model.Service) error { return f.autoErr } + func (f *fakeJobs) Trigger(_ context.Context, _ int64, by string) (model.Job, error) { f.mu.Lock() defer f.mu.Unlock() @@ -504,3 +530,114 @@ func TestJob_NotFound(t *testing.T) { assert.Equal(t, http.StatusNotFound, code) } } + +func postMode(t *testing.T, srv *httptest.Server, id string, form url.Values) (*http.Response, string) { + t.Helper() + form.Set("csrf_token", "csrf") + resp, err := noRedirectClient().PostForm(srv.URL+"/services/"+id+"/mode", form) + require.NoError(t, err) + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return resp, string(body) +} + +func TestSetMode_AutoRedirectsToServiceAndPublishes(t *testing.T) { + srv, store, _, broker, _ := newTestEnv(t) + sub, cancel := broker.Subscribe() + defer cancel() + + resp, _ := postMode(t, srv, "1", url.Values{"mode": {"auto"}}) + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "/services/1", resp.Header.Get("Location")) + assert.Equal(t, "auto", store.modeOf(1)) + assert.Equal(t, events.Services, <-sub) +} + +func TestSetMode_ReturnToOverview(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + resp, _ := postMode(t, srv, "1", url.Values{"mode": {"auto"}, "return": {"overview"}}) + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "/", resp.Header.Get("Location")) + assert.Equal(t, "auto", store.modeOf(1)) +} + +func TestSetMode_AutoRefusedShows409(t *testing.T) { + srv, store, _, _, jobs := newTestEnv(t) + jobs.autoErr = &model.NotUpdatableError{Reason: model.ReasonBuilt} + + resp, body := postMode(t, srv, "1", url.Values{"mode": {"auto"}}) + + assert.Equal(t, http.StatusConflict, resp.StatusCode) + assert.Contains(t, body, "Cannot enable auto: "+model.ReasonBuilt+".") + assert.Equal(t, "manual", store.modeOf(1)) +} + +func TestSetMode_ManualAlwaysAllowed(t *testing.T) { + srv, store, _, _, jobs := newTestEnv(t) + store.services[0].UpdateMode = "auto" + jobs.autoErr = &model.NotUpdatableError{Reason: model.ReasonBuilt} + + resp, _ := postMode(t, srv, "1", url.Values{"mode": {"manual"}}) + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "manual", store.modeOf(1)) +} + +func TestSetMode_InvalidModeAndUnknownService(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + store.services = append(store.services, model.Service{ID: 5, Kind: protocol.KindCompose, UpdateMode: "manual", RemovedAt: ptr(time.Now())}) + + resp, _ := postMode(t, srv, "1", url.Values{"mode": {"yolo"}}) + assert.Equal(t, http.StatusBadRequest, resp.StatusCode) + resp, _ = postMode(t, srv, "1", url.Values{}) + assert.Equal(t, http.StatusBadRequest, resp.StatusCode) + resp, _ = postMode(t, srv, "99", url.Values{"mode": {"auto"}}) + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + resp, _ = postMode(t, srv, "99", url.Values{"mode": {"manual"}}) + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + resp, _ = postMode(t, srv, "5", url.Values{"mode": {"auto"}}) + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + resp, _ = postMode(t, srv, "abc", url.Values{"mode": {"auto"}}) + assert.Equal(t, http.StatusBadRequest, resp.StatusCode) + assert.Equal(t, "manual", store.modeOf(1)) +} + +func TestOverview_ShowsModeToggleAndAvailableBadge(t *testing.T) { + srv, store, _, _, jobs := newTestEnv(t) + store.services[0].AvailableRef = "gitea.shcizo.se/shcizo/slusk:1.5.0" + store.services[1].UpdateMode = "auto" + jobs.autoErr = &model.NotUpdatableError{Reason: model.ReasonSelf} + + for _, path := range []string{"/", "/fragments/services"} { + _, body := get(t, srv, path) + assert.Contains(t, body, `action="/services/1/mode"`, path) + assert.Contains(t, body, `name="csrf_token" value="csrf"`, path) + assert.Contains(t, body, `name="return" value="overview"`, path) + assert.Contains(t, body, `class="badge badge-update" title="gitea.shcizo.se/shcizo/slusk:1.5.0"`, path) + assert.Contains(t, body, ``, "auto services toggle to manual: "+path) + assert.Contains(t, body, `disabled>Make auto `+model.ReasonSelf+``, "blocked manual services get a disabled button: "+path) + } +} + +func TestServicePage_ShowsAvailableAndModeForm(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + at := time.Date(2026, 10, 12, 12, 30, 0, 0, time.UTC) + store.services[0].AvailableRef = "gitea.shcizo.se/shcizo/slusk:1.5.0" + store.services[0].AvailableSource = "webhook" + store.services[0].AvailableAt = &at + + _, body := get(t, srv, "/services/1") + + assert.Contains(t, body, "gitea.shcizo.se/shcizo/slusk:1.5.0") + assert.Contains(t, body, "via webhook, 2026-10-12 12:30") + assert.Contains(t, body, `action="/services/1/mode"`) + assert.Contains(t, body, ``) +} + +func TestServicePage_AutoBlockedDisablesToggle(t *testing.T) { + srv, _, _, _, jobs := newTestEnv(t) + jobs.autoErr = &model.NotUpdatableError{Reason: model.ReasonContainer} + _, body := get(t, srv, "/services/1") + assert.Contains(t, body, "Auto unavailable: "+model.ReasonContainer+".") +} diff --git a/internal/master/web/static/app.css b/internal/master/web/static/app.css index c8d8b8c..e50510a 100644 --- a/internal/master/web/static/app.css +++ b/internal/master/web/static/app.css @@ -28,3 +28,6 @@ form.inline { display: flex; gap: 8px; margin-bottom: 16px; } table.details { width: auto; } table.details th { padding-right: 24px; } pre.log { white-space: pre-wrap; } +.badge-update { color: var(--warn); border-color: var(--warn); } +form.mode { display: inline; margin-left: 8px; } +form.mode button { font-size: 12px; padding: 0 6px; } diff --git a/internal/master/web/templates/overview.html b/internal/master/web/templates/overview.html index 8ae6cc4..3ff03b4 100644 --- a/internal/master/web/templates/overview.html +++ b/internal/master/web/templates/overview.html @@ -14,10 +14,19 @@ {{range .Services}} - {{.DisplayName}}{{if .Self}} managed outside{{end}} + {{.DisplayName}}{{if .Self}} managed outside{{end}}{{if .AvailableRef}} update available{{end}} {{shortRef .ActualRef}} {{.Status}} - {{.UpdateMode}} + {{.UpdateMode}} +
+ + + {{if eq .UpdateMode "auto"}} + + {{else}} + {{if .AutoBlocked}} {{.AutoBlocked}}{{end}}{{end}} +
+ {{end}} diff --git a/internal/master/web/templates/service.html b/internal/master/web/templates/service.html index fe58b01..68c0082 100644 --- a/internal/master/web/templates/service.html +++ b/internal/master/web/templates/service.html @@ -8,6 +8,8 @@ {{if .Service.ActualDigest}}Digest{{.Service.ActualDigest}}{{end}} Status{{.Service.Status}} {{if .Service.DesiredRef}}Desired image{{shortRef .Service.DesiredRef}}{{end}} + {{if .Service.AvailableRef}}Available{{.Service.AvailableRef}} via {{.Service.AvailableSource}}{{with .Service.AvailableAt}}, {{.Format "2006-01-02 15:04"}}{{end}}{{end}} + Mode{{.Service.UpdateMode}} {{if .ConfigFiles}}Config files{{range .ConfigFiles}}{{.}}
{{end}}{{end}} {{if .Reason}} @@ -19,6 +21,16 @@

Pulls the current tag again and redeploys the service.

{{end}} +
+ + {{if eq .Service.UpdateMode "auto"}} + +

The webhook will only mark new versions as available.

+ {{else}} + +

{{if .AutoBlocked}}Auto unavailable: {{.AutoBlocked}}.{{else}}The webhook will update this service as soon as CI pushes a new image.{{end}}

+ {{end}} +

Update history

{{template "history" .}} diff --git a/internal/master/web/views.go b/internal/master/web/views.go index 824100c..361d7e1 100644 --- a/internal/master/web/views.go +++ b/internal/master/web/views.go @@ -11,12 +11,20 @@ import ( type serviceGroup struct { Title string - Services []model.Service + Services []serviceRow +} + +// serviceRow is a service as shown in the overview table. +type serviceRow struct { + model.Service + AutoBlocked string // why the service cannot be set to auto; "" when it can } // groupServices groups machine-bound services per agent and Swarm services // in one group, keeping the store's order (machines first, Swarm last). -func groupServices(services []model.Service) []serviceGroup { +// autoBlocked is called for every service and returns the reason it cannot be +// set to auto, or "" when it can. +func groupServices(services []model.Service, autoBlocked func(model.Service) string) []serviceGroup { var groups []serviceGroup index := map[string]int{} for _, s := range services { @@ -32,7 +40,7 @@ func groupServices(services []model.Service) []serviceGroup { index[title] = i groups = append(groups, serviceGroup{Title: title}) } - groups[i].Services = append(groups[i].Services, s) + groups[i].Services = append(groups[i].Services, serviceRow{Service: s, AutoBlocked: autoBlocked(s)}) } return groups } diff --git a/internal/master/webhook/webhook.go b/internal/master/webhook/webhook.go new file mode 100644 index 0000000..36ebfbf --- /dev/null +++ b/internal/master/webhook/webhook.go @@ -0,0 +1,350 @@ +// Package webhook implements POST /update, the endpoint CI calls after +// pushing an image. It records the pushed tag as "available" on every service +// that uses the image, starts an update for the services in auto mode and +// waits for those updates to finish. +package webhook + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "encoding/json" + "errors" + "log/slog" + "net/http" + "regexp" + "strconv" + "strings" + "time" + + "github.com/shcizo/package-updater/internal/imageref" + "github.com/shcizo/package-updater/internal/logging" + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" +) + +// maxBodyBytes caps the request body; a valid request is a few hundred bytes. +const maxBodyBytes = 64 << 10 + +// defaultPollEvery is how often pending jobs are re-read from the store when +// no broker signal arrives (signals to a slow subscriber can be dropped). +const defaultPollEvery = 5 * time.Second + +// finalReadTimeout bounds the last read of pending jobs after the wait ends. +const finalReadTimeout = 2 * time.Second + +// tagPattern is Docker's tag grammar. It keeps spaces, colons and "@" out of +// the ref the agent is asked to deploy. +var tagPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$`) + +// requestIDPattern limits a caller-supplied X-Request-ID to what is safe to +// echo in a header and write to the log. +var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9._-]{1,128}$`) + +// Result statuses. +const ( + statusSucceeded = "succeeded" + statusFailed = "failed" + statusTimeout = "timeout" + statusMarked = "marked" + statusBusy = "busy" + statusRefused = "refused" +) + +// Store is the storage the webhook needs. *store.Store satisfies it. +type Store interface { + ServicesByRepo(ctx context.Context, repo string) ([]model.Service, error) + SetAvailable(ctx context.Context, id int64, a model.Available) error + GetJob(ctx context.Context, id int64) (model.Job, error) +} + +// Jobs starts updates. *jobs.Service satisfies it. +type Jobs interface { + Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) +} + +// Broker delivers change signals. *events.Broker satisfies it. +type Broker interface { + Subscribe() (<-chan events.Topic, func()) + Publish(t events.Topic) +} + +// Config holds the webhook settings. +type Config struct { + Token string // bearer token; empty disables the webhook + Timeout time.Duration // bounds the whole wait of one request + PublicURL string // base of job_url, without trailing slash +} + +// Handler serves POST /update. +type Handler struct { + store Store + jobs Jobs + broker Broker + cfg Config + log *slog.Logger + + now func() time.Time + pollEvery time.Duration +} + +// New returns the handler for POST /update. It does its own bearer-token +// check, so mount it outside the OIDC login. +func New(store Store, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) *Handler { + return &Handler{store: store, jobs: jobs, broker: broker, cfg: cfg, log: log, + now: time.Now, pollEvery: defaultPollEvery} +} + +type request struct { + Image string `json:"image"` + Tag string `json:"tag"` +} + +type result struct { + ServiceID int64 `json:"service_id"` + Service string `json:"service"` + Kind string `json:"kind"` + Mode string `json:"mode"` + Status string `json:"status"` + JobID int64 `json:"job_id,omitempty"` + JobURL string `json:"job_url,omitempty"` + ToRef string `json:"to_ref,omitempty"` + Error string `json:"error,omitempty"` + DurationMS int64 `json:"duration_ms"` +} + +type response struct { + RequestID string `json:"request_id"` + Image string `json:"image"` + Tag string `json:"tag,omitempty"` + Matched int `json:"matched"` + Results []result `json:"results"` +} + +// ServeHTTP handles one webhook call. See the README for the contract. +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + started := h.now() + requestID := r.Header.Get("X-Request-ID") + if !requestIDPattern.MatchString(requestID) { + requestID = newRequestID() + } + w.Header().Set("X-Request-ID", requestID) + + if h.cfg.Token == "" { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "webhook disabled: UPDATER_API_KEY not set"}) + return + } + if !h.authorized(r) { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + + var req request + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON body"}) + return + } + if req.Image == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "image is required"}) + return + } + if req.Tag != "" && !tagPattern.MatchString(req.Tag) { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid tag"}) + return + } + + log := logging.FromContext(logging.WithRequestID(r.Context(), requestID), h.log) + repo := imageref.Parse(req.Image).Repo + // A client that disconnects must not abort marking or starting: that + // could leave a queued job no agent knows about. + work := context.WithoutCancel(r.Context()) + + services, err := h.store.ServicesByRepo(work, repo) + if err != nil { + log.Error("webhook_lookup_failed", "image", req.Image, "error", err) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal error"}) + return + } + if len(services) == 0 { + writeJSON(w, http.StatusNotFound, map[string]string{ + "error": "no service uses image " + repo, "request_id": requestID}) + return + } + + // Subscribe before any job exists so no completion signal is missed. + sub, cancel := h.broker.Subscribe() + defer cancel() + + resp := response{RequestID: requestID, Image: req.Image, Tag: req.Tag, Matched: len(services)} + pending := map[int]int64{} // index into resp.Results -> job ID + for _, svc := range services { + res, job := h.handle(work, svc, req.Tag, started, log) + resp.Results = append(resp.Results, res) + if job != nil { + pending[len(resp.Results)-1] = job.ID + } + } + h.broker.Publish(events.Services) + + h.wait(r.Context(), sub, &resp, pending, started, log) + + status := http.StatusOK + counts := map[string]int{} + for _, res := range resp.Results { + counts[res.Status]++ + if res.Status == statusFailed || res.Status == statusTimeout { + status = http.StatusInternalServerError + } + } + log.Info("webhook_request", "image", req.Image, "tag", req.Tag, "matched", resp.Matched, + "succeeded", counts[statusSucceeded], "failed", counts[statusFailed], "timeout", counts[statusTimeout], + "marked", counts[statusMarked], "busy", counts[statusBusy], "refused", counts[statusRefused]) + writeJSON(w, status, resp) +} + +// handle records the available ref on svc and, in auto mode, starts an update. +// It returns the result and, if the job is still active, that job. +func (h *Handler) handle(ctx context.Context, svc model.Service, tag string, started time.Time, log *slog.Logger) (result, *model.Job) { + res := result{ServiceID: svc.ID, Service: svc.DisplayName(), Kind: string(svc.Kind), Mode: svc.UpdateMode} + + // The ref is built from the service's own repo, never from the request + // string, so a request cannot point a service at another registry. + var ref string + if tag != "" { + ref = svc.ImageRepo + ":" + tag + } else { + ref = jobs.FollowedRef(svc.ActualRef) + } + if ref == "" { + return h.finish(res, statusRefused, "no tag given and the service follows no tag", started), nil + } + res.ToRef = ref + err := h.store.SetAvailable(ctx, svc.ID, model.Available{Ref: ref, Source: model.SourceWebhook, At: h.now()}) + if err != nil { + log.Error("webhook_set_available_failed", "service_id", svc.ID, "error", err) + return h.finish(res, statusFailed, "could not record available version", started), nil + } + if svc.UpdateMode != model.ModeAuto { + return h.finish(res, statusMarked, "", started), nil + } + + job, err := h.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerWebhook, Ref: ref}) + var notUpdatable *model.NotUpdatableError + switch { + case errors.As(err, ¬Updatable): + return h.finish(res, statusRefused, notUpdatable.Reason, started), nil + case errors.Is(err, model.ErrJobActive): + h.setJob(&res, job) + return h.finish(res, statusBusy, err.Error(), started), nil + case err != nil: + log.Error("webhook_start_failed", "service_id", svc.ID, "error", err) + return h.finish(res, statusFailed, "could not start job", started), nil + } + h.setJob(&res, job) + if job.Active() { + return res, &job + } + h.resolveJob(&res, job, started) + return res, nil +} + +func (h *Handler) setJob(res *result, job model.Job) { + res.JobID = job.ID + res.JobURL = h.cfg.PublicURL + "/jobs/" + strconv.FormatInt(job.ID, 10) +} + +func (h *Handler) finish(res result, status, msg string, started time.Time) result { + res.Status, res.Error = status, msg + res.DurationMS = h.now().Sub(started).Milliseconds() + return res +} + +// resolveJob sets the result from a job that is no longer active. +func (h *Handler) resolveJob(res *result, job model.Job, started time.Time) { + if job.Status == model.JobSucceeded { + *res = h.finish(*res, statusSucceeded, "", started) + return + } + *res = h.finish(*res, statusFailed, job.Error, started) +} + +// wait blocks until every pending job has finished, the timeout passes or +// the client goes away. Jobs still pending then get status "timeout"; the +// master never cancels or fails them. +func (h *Handler) wait(parent context.Context, sub <-chan events.Topic, resp *response, + pending map[int]int64, started time.Time, log *slog.Logger) { + if len(pending) == 0 { + return + } + ctx, cancel := context.WithTimeout(parent, h.cfg.Timeout) + defer cancel() + ticker := time.NewTicker(h.pollEvery) + defer ticker.Stop() + + for len(pending) > 0 { + select { + case <-ctx.Done(): + // A job may have finished while the timer fired: look once more, + // on a context that outlives the request. + final, stop := context.WithTimeout(context.WithoutCancel(parent), finalReadTimeout) + h.readPending(final, resp, pending, started, log) + stop() + for i := range pending { + resp.Results[i] = h.finish(resp.Results[i], statusTimeout, "still running when the request timed out", started) + } + return + case t := <-sub: + if t != events.Jobs { + continue + } + case <-ticker.C: + } + h.readPending(ctx, resp, pending, started, log) + } +} + +// readPending re-reads every pending job and resolves the ones that are done. +func (h *Handler) readPending(ctx context.Context, resp *response, pending map[int]int64, + started time.Time, log *slog.Logger) { + for i, id := range pending { + job, err := h.store.GetJob(ctx, id) + if err != nil { + log.Warn("webhook_get_job_failed", "job_id", id, "error", err) + continue + } + if !job.Active() { + h.resolveJob(&resp.Results[i], job, started) + delete(pending, i) + } + } +} + +// authorized reports whether the request carries the configured bearer token. +// Both sides are hashed so the comparison takes the same time whatever the +// length of the supplied token. +func (h *Handler) authorized(r *http.Request) bool { + const prefix = "Bearer " + got, ok := strings.CutPrefix(r.Header.Get("Authorization"), prefix) + if !ok || got == "" { + return false + } + a, b := sha256.Sum256([]byte(got)), sha256.Sum256([]byte(h.cfg.Token)) + return subtle.ConstantTimeCompare(a[:], b[:]) == 1 +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +// newRequestID returns a random 128-bit hex ID. +func newRequestID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} diff --git a/internal/master/webhook/webhook_test.go b/internal/master/webhook/webhook_test.go new file mode 100644 index 0000000..57fde56 --- /dev/null +++ b/internal/master/webhook/webhook_test.go @@ -0,0 +1,517 @@ +package webhook + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +const token = "s3cret" + +type availCall struct { + id int64 + a model.Available +} + +type fakeStore struct { + mu sync.Mutex + services []model.Service // returned for any repo that equals their ImageRepo + jobs map[int64]model.Job + avail []availCall + repos []string +} + +func (f *fakeStore) ServicesByRepo(_ context.Context, repo string) ([]model.Service, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.repos = append(f.repos, repo) + var out []model.Service + for _, s := range f.services { + if s.ImageRepo == repo { + out = append(out, s) + } + } + return out, nil +} + +func (f *fakeStore) SetAvailable(_ context.Context, id int64, a model.Available) error { + f.mu.Lock() + defer f.mu.Unlock() + f.avail = append(f.avail, availCall{id, a}) + return nil +} + +func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.jobs[id], nil +} + +func (f *fakeStore) setJob(j model.Job) { + f.mu.Lock() + defer f.mu.Unlock() + f.jobs[j.ID] = j +} + +type startCall struct { + svc model.Service + req jobs.StartRequest + ctx context.Context +} + +// fakeJobs hands out job IDs and applies per-service behaviour. +type fakeJobs struct { + mu sync.Mutex + store *fakeStore + calls []startCall + errFor map[int64]error // returned together with errJob[id] if set + jobFor map[int64]model.Job // status of the job returned by Start + next int64 +} + +func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, startCall{svc, req, ctx}) + if err, ok := f.errFor[svc.ID]; ok { + return f.jobFor[svc.ID], err + } + f.next++ + j := model.Job{ID: f.next, ServiceID: svc.ID, Status: model.JobQueued, Trigger: req.Trigger, ToRef: req.Ref} + if o, ok := f.jobFor[svc.ID]; ok { + o.ID = j.ID + j = o + } + f.store.mu.Lock() + f.store.jobs[j.ID] = j + f.store.mu.Unlock() + return j, nil +} + +func (f *fakeJobs) startCalls() []startCall { + f.mu.Lock() + defer f.mu.Unlock() + return append([]startCall(nil), f.calls...) +} + +type env struct { + store *fakeStore + jobs *fakeJobs + broker *events.Broker + h *Handler +} + +func newEnv(t *testing.T, services ...model.Service) *env { + t.Helper() + st := &fakeStore{services: services, jobs: map[int64]model.Job{}} + fj := &fakeJobs{store: st, errFor: map[int64]error{}, jobFor: map[int64]model.Job{}} + b := events.NewBroker() + h := New(st, fj, b, Config{Token: token, Timeout: 2 * time.Second, PublicURL: "https://updater.example.com"}, + slog.New(slog.NewTextHandler(io.Discard, nil))) + h.now = func() time.Time { return time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) } + return &env{store: st, jobs: fj, broker: b, h: h} +} + +func (e *env) post(body string, headers ...string) *httptest.ResponseRecorder { + return e.postCtx(context.Background(), body, headers...) +} + +func (e *env) postCtx(ctx context.Context, body string, headers ...string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(body)).WithContext(ctx) + req.Header.Set("Authorization", "Bearer "+token) + for i := 0; i+1 < len(headers); i += 2 { + req.Header.Set(headers[i], headers[i+1]) + } + rec := httptest.NewRecorder() + e.h.ServeHTTP(rec, req) + return rec +} + +func decode(t *testing.T, rec *httptest.ResponseRecorder) response { + t.Helper() + var r response + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &r), rec.Body.String()) + return r +} + +func compose(id int64, mode, actualRef string) model.Service { + return model.Service{ID: id, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web" + string(rune('a'+id)), + ImageRepo: "gitea.shcizo.se/shcizo/slusk", ActualRef: actualRef, UpdateMode: mode} +} + +const slusk = "gitea.shcizo.se/shcizo/slusk" + +func TestAuth(t *testing.T) { + tests := []struct { + name string + serverKey string + header string + wantStatus int + wantBody string + }{ + {"missing header", token, "", 401, `{"error":"unauthorized"}`}, + {"wrong scheme", token, "Basic " + token, 401, `{"error":"unauthorized"}`}, + {"wrong token", token, "Bearer nope", 401, `{"error":"unauthorized"}`}, + {"empty bearer", token, "Bearer ", 401, `{"error":"unauthorized"}`}, + {"key unset", "", "Bearer " + token, 503, `{"error":"webhook disabled: UPDATER_API_KEY not set"}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + e.h.cfg.Token = tt.serverKey + req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"`+slusk+`"}`)) + if tt.header != "" { + req.Header.Set("Authorization", tt.header) + } + rec := httptest.NewRecorder() + e.h.ServeHTTP(rec, req) + assert.Equal(t, tt.wantStatus, rec.Code) + assert.JSONEq(t, tt.wantBody, rec.Body.String()) + assert.Empty(t, e.store.repos, "nothing may be looked up before auth") + }) + } +} + +func TestBadRequest(t *testing.T) { + tests := []struct { + name, body, wantError string + }{ + {"invalid JSON", `{`, "invalid JSON body"}, + {"empty body", ``, "invalid JSON body"}, + {"empty image", `{"image":""}`, "image is required"}, + {"missing image", `{"tag":"v1"}`, "image is required"}, + {"oversized", `{"image":"` + strings.Repeat("a", 70<<10) + `"}`, "invalid JSON body"}, + {"tag with digest", `{"image":"` + slusk + `","tag":"v1@sha256:abc"}`, "invalid tag"}, + {"tag with space", `{"image":"` + slusk + `","tag":"v 1"}`, "invalid tag"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1")) + rec := e.post(tt.body) + assert.Equal(t, 400, rec.Code) + assert.JSONEq(t, `{"error":"`+tt.wantError+`"}`, rec.Body.String()) + assert.Empty(t, e.jobs.startCalls()) + }) + } +} + +func TestNoMatch(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1")) + for _, image := range []string{"gitea.shcizo.se/shcizo/other", "gitea.shcizo.se/shcizo/Slusk"} { + rec := e.post(`{"image":"` + image + `"}`) + assert.Equal(t, 404, rec.Code, image) + assert.Contains(t, rec.Body.String(), "no service uses image "+image) + } + assert.Empty(t, e.jobs.startCalls()) + assert.Empty(t, e.store.avail) +} + +func TestImageIsNormalisedBeforeMatching(t *testing.T) { + svc := compose(1, model.ModeManual, slusk+":1") + local := compose(2, model.ModeManual, "localhost:5000/foo:v1") + local.ImageRepo = "localhost:5000/foo" + for _, tt := range []struct{ image, wantRepo string }{ + {slusk, slusk}, + {slusk + ":2.0", slusk}, + {slusk + "@sha256:abc", slusk}, + {slusk + ":2.0@sha256:abc", slusk}, + {"localhost:5000/foo:v1", "localhost:5000/foo"}, + } { + e := newEnv(t, svc, local) + rec := e.post(`{"image":"` + tt.image + `"}`) + assert.Equal(t, 200, rec.Code, tt.image) + assert.Equal(t, []string{tt.wantRepo}, e.store.repos, tt.image) + assert.Equal(t, 1, decode(t, rec).Matched, tt.image) + } +} + +func TestManualServiceIsMarkedNotUpdated(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1.4.2@sha256:old")) + sub, cancel := e.broker.Subscribe() + defer cancel() + + rec := e.post(`{"image":"` + slusk + `:ignored-in-ref","tag":"1.5.0"}`) + + assert.Equal(t, 200, rec.Code) + assert.Empty(t, e.jobs.startCalls()) + require.Len(t, e.store.avail, 1) + assert.Equal(t, int64(1), e.store.avail[0].id) + assert.Equal(t, model.Available{Ref: slusk + ":1.5.0", Source: model.SourceWebhook, At: e.h.now()}, e.store.avail[0].a) + resp := decode(t, rec) + assert.Equal(t, 1, resp.Matched) + require.Len(t, resp.Results, 1) + r := resp.Results[0] + assert.Equal(t, statusMarked, r.Status) + assert.Equal(t, "slusk/webb", r.Service) + assert.Equal(t, "compose", r.Kind) + assert.Equal(t, "manual", r.Mode) + assert.Equal(t, slusk+":1.5.0", r.ToRef) + assert.Zero(t, r.JobID) + assert.NotEmpty(t, resp.RequestID) + assert.Equal(t, resp.RequestID, rec.Header().Get("X-Request-ID")) + assert.Equal(t, events.Services, <-sub) +} + +func TestRequestIDIsEchoed(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + rec := e.post(`{"image":"`+slusk+`"}`, "X-Request-ID", "ci-123") + assert.Equal(t, "ci-123", decode(t, rec).RequestID) + assert.Equal(t, "ci-123", rec.Header().Get("X-Request-ID")) +} + +func TestAvailableRefWithoutTag(t *testing.T) { + tests := []struct { + name string + actualRef string + wantRef string // "" means SetAvailable must not be called + }{ + {"followed tag", slusk + ":1.4.2@sha256:old", slusk + ":1.4.2"}, + {"bare image follows latest", slusk, slusk + ":latest"}, + {"digest only", slusk + "@sha256:old", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, tt.actualRef)) + rec := e.post(`{"image":"` + slusk + `"}`) + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + if tt.wantRef == "" { + assert.Empty(t, e.store.avail) + assert.Equal(t, statusRefused, r.Status) + assert.Equal(t, "no tag given and the service follows no tag", r.Error) + return + } + assert.Equal(t, statusMarked, r.Status) + require.Len(t, e.store.avail, 1) + assert.Equal(t, tt.wantRef, e.store.avail[0].a.Ref) + }) + } +} + +func TestAutoDigestOnlyServiceWithoutTagIsRefused(t *testing.T) { + svc := compose(1, model.ModeAuto, slusk+"@sha256:old") + e := newEnv(t, svc) + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + assert.Empty(t, e.store.avail) + assert.Empty(t, e.jobs.startCalls(), "no target ref: nothing to start") + r := decode(t, rec).Results[0] + assert.Equal(t, statusRefused, r.Status) + assert.Equal(t, "no tag given and the service follows no tag", r.Error) +} + +func TestValidTags(t *testing.T) { + for _, tag := range []string{"latest", "1.2.3-rc.1", "v1_2", strings.Repeat("a", 128)} { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + rec := e.post(`{"image":"` + slusk + `","tag":"` + tag + `"}`) + assert.Equal(t, 200, rec.Code, tag) + require.Len(t, e.store.avail, 1, tag) + assert.Equal(t, slusk+":"+tag, e.store.avail[0].a.Ref) + } + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + rec := e.post(`{"image":"` + slusk + `","tag":"` + strings.Repeat("a", 129) + `"}`) + assert.Equal(t, 400, rec.Code) +} + +func TestUnsafeRequestIDIsReplaced(t *testing.T) { + for _, id := range []string{"has space", "bad/slash", strings.Repeat("a", 129)} { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + rec := e.post(`{"image":"`+slusk+`"}`, "X-Request-ID", id) + got := rec.Header().Get("X-Request-ID") + assert.NotEqual(t, id, got) + assert.Regexp(t, `^[0-9a-f]{32}$`, got) + } + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + rec := e.post(`{"image":"`+slusk+`"}`, "X-Request-ID", strings.Repeat("a", 128)) + assert.Equal(t, strings.Repeat("a", 128), rec.Header().Get("X-Request-ID")) +} + +func TestJobFinishedJustBeforeTimeoutIsNotReportedAsTimeout(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.cfg.Timeout = 200 * time.Millisecond + e.h.pollEvery = time.Hour // only the final read after the deadline can see the result + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded}) // no Publish + }() + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + assert.Equal(t, statusSucceeded, decode(t, rec).Results[0].Status) +} + +func TestAutoJobTargetIsBuiltFromServiceRepo(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2@sha256:old")) + e.jobs.jobFor[1] = model.Job{Status: model.JobSucceeded} + + rec := e.post(`{"image":"` + slusk + `:other","tag":"2.0.0"}`) + + assert.Equal(t, 200, rec.Code) + calls := e.jobs.startCalls() + require.Len(t, calls, 1) + assert.Equal(t, slusk+":2.0.0", calls[0].req.Ref) + assert.Equal(t, model.TriggerWebhook, calls[0].req.Trigger) + assert.Empty(t, calls[0].req.RequestedBy) + require.Len(t, e.store.avail, 1) + assert.Equal(t, slusk+":2.0.0", e.store.avail[0].a.Ref) +} + +func TestAutoJobSucceeded(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded}) + e.broker.Publish(events.JobTopic(1)) // progress topic is ignored + e.broker.Publish(events.Jobs) + }() + rec := e.post(`{"image":"` + slusk + `","tag":"1.5.0"}`) + + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusSucceeded, r.Status) + assert.Equal(t, int64(1), r.JobID) + assert.Equal(t, "https://updater.example.com/jobs/1", r.JobURL) + assert.Equal(t, "auto", r.Mode) +} + +func TestAutoJobFailedIs500(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobFailed, Error: "pull access denied"}) + e.broker.Publish(events.Jobs) + }() + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 500, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusFailed, r.Status) + assert.Equal(t, "pull access denied", r.Error) +} + +func TestJobFailedAtStartIsNotWaitedFor(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.jobs.jobFor[1] = model.Job{Status: model.JobFailed, Error: "agent offline"} + e.h.cfg.Timeout = time.Minute + + begin := time.Now() + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Less(t, time.Since(begin), 5*time.Second) + assert.Equal(t, 500, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusFailed, r.Status) + assert.Equal(t, "agent offline", r.Error) + assert.NotZero(t, r.JobID) +} + +func TestTimeout(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.cfg.Timeout = 50 * time.Millisecond + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 500, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusTimeout, r.Status) + assert.Equal(t, int64(1), r.JobID) + assert.Equal(t, model.JobQueued, e.store.jobs[1].Status, "the master must not touch a timed-out job") +} + +func TestDroppedSignalIsRecoveredByPolling(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.pollEvery = 5 * time.Millisecond + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded}) // no Publish + }() + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + assert.Equal(t, statusSucceeded, decode(t, rec).Results[0].Status) +} + +func TestBusyIsNotAFailure(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.jobs.errFor[1] = model.ErrJobActive + e.jobs.jobFor[1] = model.Job{ID: 42, Status: model.JobRunning} + e.h.cfg.Timeout = time.Minute + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusBusy, r.Status) + assert.Equal(t, int64(42), r.JobID) + assert.Equal(t, "https://updater.example.com/jobs/42", r.JobURL) + assert.NotEmpty(t, e.store.avail, "the available ref is still recorded") +} + +func TestRefusedIsNotAFailure(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.jobs.errFor[1] = &model.NotUpdatableError{Reason: model.ReasonSelf} + + rec := e.post(`{"image":"` + slusk + `","tag":"2"}`) + + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusRefused, r.Status) + assert.Equal(t, model.ReasonSelf, r.Error) + assert.Zero(t, r.JobID) + assert.NotEmpty(t, e.store.avail) +} + +func TestMixedAutoAndManual(t *testing.T) { + e := newEnv(t, + compose(1, model.ModeManual, slusk+":1"), + compose(2, model.ModeAuto, slusk+":1"), + compose(3, model.ModeAuto, slusk+":1")) + e.jobs.jobFor[2] = model.Job{Status: model.JobSucceeded} + e.jobs.jobFor[3] = model.Job{Status: model.JobFailed, Error: "boom"} + + rec := e.post(`{"image":"` + slusk + `","tag":"2"}`) + + assert.Equal(t, 500, rec.Code) + resp := decode(t, rec) + assert.Equal(t, 3, resp.Matched) + got := []string{resp.Results[0].Status, resp.Results[1].Status, resp.Results[2].Status} + assert.Equal(t, []string{statusMarked, statusSucceeded, statusFailed}, got) + assert.Len(t, e.store.avail, 3) +} + +func TestClientCancelStopsWaitingButNotStarting(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.cfg.Timeout = time.Minute + ctx, cancel := context.WithCancel(context.Background()) + cancel() // the client is already gone when the handler runs + + begin := time.Now() + rec := e.postCtx(ctx, `{"image":"`+slusk+`"}`) + + assert.Less(t, time.Since(begin), 5*time.Second) + assert.Equal(t, statusTimeout, decode(t, rec).Results[0].Status) + calls := e.jobs.startCalls() + require.Len(t, calls, 1, "the job must still be started") + assert.NoError(t, calls[0].ctx.Err(), "Start must run on a context the client cannot cancel") +}