diff --git a/CLAUDE.md b/CLAUDE.md index 997ab0f..c9a3c8d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -198,6 +198,7 @@ See `config-appview.example.yaml` and `config-hold.example.yaml` for all options - **Hold stats are ATProto records in CAR store** — `io.atcr.hold.stats` records are stored via `repomgr.PutRecord()`, not in SQLite. Lost if CAR store is lost without backup. - **PLC auto-update on boot** — When using did:plc, `LoadOrCreateDID()` calls `EnsurePLCCurrent()` every startup. If local signing key or URL doesn't match plc.directory, it auto-updates (requires rotation key on disk). - **Hold CAR store is the source of truth** — Captain, crew, layer, stats, scan records, Bluesky posts, profiles are all ATProto records in the CAR store. SQLite holds only the records index and events. +- **Admin panel long-running loops MUST use background jobs** — Any admin handler that loops over many items doing per-item PDS writes/network calls (bulk crew tier remap, crew import, scan backfill) must NOT run synchronously on the request context. The reverse proxy 504s such a request around the 10s mark and cancels `r.Context()`, aborting the loop mid-flight (`getb tx, context canceled`). Use the `startJob` helper in `pkg/hold/admin/jobs.go`: do fast validation synchronously (return a 200 fragment on failure, never a 302 — the forms are htmx-driven), then kick off the loop via `ui.startJob(key, title, resultTemplate, timeout, fn)` and render `partials/job_progress.html`. The fragment polls `/admin/api/jobs/{key}/status` until done. `fn` runs under its own detached `context.Background()` timeout, so it survives the request ending. The GC subsystem (`gc.startBackground`) is a peer implementation in the `gc` package (which must not import `admin`). ## Common Tasks diff --git a/Makefile b/Makefile index f02526b..32d689b 100644 --- a/Makefile +++ b/Makefile @@ -23,8 +23,13 @@ GENERATED_ASSETS = \ pkg/appview/public/js/lucide.min.js \ pkg/appview/licenses/spdx-licenses.json -generate: $(GENERATED_ASSETS) ## Run go generate to download vendor assets +generate: ## Run go generate ./... (always — regenerates cbor_gen, icon sprites, vendor assets) + @echo "→ Running go generate ./..." + go generate ./... +# File rule: lazily download missing vendor assets for fast incremental local builds. +# Production builds depend on the phony `generate` target instead so generated code +# (cbor_gen.go, icon sprites, etc.) is always up to date. $(GENERATED_ASSETS): @echo "→ Generating vendor assets and code..." go generate ./... @@ -66,7 +71,7 @@ build-oauth-helper: ## Build OAuth helper only # 2.43, which otherwise stamps sqrtf@GLIBC_2.43 onto cgo-linked output). TRIXIE_BUILDER_IMAGE ?= golang:1-trixie -build-trixie: $(GENERATED_ASSETS) ## Build all production binaries (appview, hold, credential-helper, scanner, labeler) for linux/amd64 in a Debian 13 (glibc 2.41) container +build-trixie: generate ## Build all production binaries (appview, hold, credential-helper, scanner, labeler) for linux/amd64 in a Debian 13 (glibc 2.41) container @echo "→ Building in $(TRIXIE_BUILDER_IMAGE) for glibc 2.41 compatibility..." @mkdir -p bin docker run --rm \ diff --git a/cmd/firehose-tap/main.go b/cmd/firehose-tap/main.go new file mode 100644 index 0000000..50110dd --- /dev/null +++ b/cmd/firehose-tap/main.go @@ -0,0 +1,395 @@ +// firehose-tap subscribes to an ATProto com.atproto.sync.subscribeRepos endpoint +// and pretty-prints every event it sees, with extra emphasis on Sync 1.1 fields +// (per-op prev CIDs, commit-level prevData) so you can verify a hold or relay +// is emitting compliant events. +// +// Typical use: +// +// # Watch new events as you trigger pushes/deletes on production +// go run ./cmd/firehose-tap wss://hold01.atcr.io +// +// # Replay from the beginning of the persisted buffer +// go run ./cmd/firehose-tap --cursor 0 wss://hold01.atcr.io +// +// # CI mode: subscribe for 30s, exit non-zero if any update/delete op is +// # missing prev (Sync 1.1 inductive-firehose requirement). +// go run ./cmd/firehose-tap --validate --duration 30s wss://hold01.atcr.io +// +// The endpoint may be a single hold or a relay. The wss:// scheme is required. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "net/http" + "net/url" + "os" + "os/signal" + "strings" + "sync/atomic" + "syscall" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/events" + "github.com/bluesky-social/indigo/events/schedulers/sequential" + "github.com/gorilla/websocket" +) + +// ANSI color codes (disabled via --no-color or NO_COLOR env) +var ( + cRed = "\033[31m" + cGreen = "\033[32m" + cYellow = "\033[33m" + cCyan = "\033[36m" + cBold = "\033[1m" + cDim = "\033[2m" + cReset = "\033[0m" +) + +func disableColors() { + cRed, cGreen, cYellow, cCyan, cBold, cDim, cReset = "", "", "", "", "", "", "" +} + +type config struct { + endpoint string + cursor int64 + cursorSet bool + didFilter string + validate bool + duration time.Duration + maxEvents int64 + verbose bool + noColor bool +} + +// violationCounts tracks Sync 1.1 compliance issues seen so far. +type violationCounts struct { + missingPrev atomic.Int64 // update/delete op without prev + createWithPrev atomic.Int64 // create op with non-nil prev (spec forbids) + deleteWithCid atomic.Int64 // delete op with non-nil cid (spec forbids) + missingPrevData atomic.Int64 // commit with since but no prevData +} + +func (v *violationCounts) total() int64 { + return v.missingPrev.Load() + v.createWithPrev.Load() + v.deleteWithCid.Load() + v.missingPrevData.Load() +} + +func main() { + cfg := parseFlags() + if cfg.noColor || os.Getenv("NO_COLOR") != "" { + disableColors() + } + + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + if cfg.duration > 0 { + var timeoutCancel context.CancelFunc + ctx, timeoutCancel = context.WithTimeout(ctx, cfg.duration) + defer timeoutCancel() + } + + if err := run(ctx, cfg); err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + fmt.Fprintf(os.Stderr, "%sfirehose-tap: %v%s\n", cRed, err, cReset) + os.Exit(2) + } +} + +func parseFlags() *config { + cfg := &config{} + flag.Int64Var(&cfg.cursor, "cursor", -1, "subscribeRepos cursor (-1 = new events only, 0 = replay from start)") + flag.StringVar(&cfg.didFilter, "did", "", "only show events for this DID (case-sensitive match on commit.repo)") + flag.BoolVar(&cfg.validate, "validate", false, "exit non-zero if any Sync 1.1 violation is observed") + flag.DurationVar(&cfg.duration, "duration", 0, "exit cleanly after this duration (e.g. 30s)") + flag.Int64Var(&cfg.maxEvents, "max-events", 0, "exit cleanly after seeing this many events (0 = unlimited)") + flag.BoolVar(&cfg.verbose, "verbose", false, "print full ops for create/no-prev-needed ops too") + flag.BoolVar(&cfg.noColor, "no-color", false, "disable ANSI colors") + + // Detect whether --cursor was explicitly set so we can omit it from the URL + // when the user didn't pass one (default -1 means "live, no backfill"). + flag.Usage = func() { + fmt.Fprintf(os.Stderr, "Usage: %s [flags] \n\n", os.Args[0]) + fmt.Fprintln(os.Stderr, "Endpoint examples:") + fmt.Fprintln(os.Stderr, " wss://hold01.atcr.io") + fmt.Fprintln(os.Stderr, " wss://relay1.us-east.bsky.network") + fmt.Fprintln(os.Stderr, " ws://localhost:8080") + fmt.Fprintln(os.Stderr) + fmt.Fprintln(os.Stderr, "Flags:") + flag.PrintDefaults() + } + flag.Parse() + + flag.Visit(func(f *flag.Flag) { + if f.Name == "cursor" { + cfg.cursorSet = true + } + }) + + args := flag.Args() + if len(args) != 1 { + flag.Usage() + os.Exit(2) + } + cfg.endpoint = args[0] + return cfg +} + +func run(ctx context.Context, cfg *config) error { + wsURL, err := buildWSURL(cfg.endpoint, cfg.cursor, cfg.cursorSet) + if err != nil { + return err + } + + fmt.Fprintf(os.Stderr, "%sfirehose-tap%s connecting to %s%s%s\n", cBold, cReset, cCyan, wsURL, cReset) + if cfg.validate { + fmt.Fprintf(os.Stderr, "%svalidate mode: process will exit non-zero on Sync 1.1 violations%s\n", cDim, cReset) + } + + header := http.Header{} + header.Set("User-Agent", "atcr-firehose-tap/1.0") + + dialer := websocket.DefaultDialer + dialer.HandshakeTimeout = 30 * time.Second + conn, resp, err := dialer.DialContext(ctx, wsURL, header) + if err != nil { + if resp != nil { + return fmt.Errorf("dial %s: %w (HTTP %d)", wsURL, err, resp.StatusCode) + } + return fmt.Errorf("dial %s: %w", wsURL, err) + } + defer conn.Close() + + v := &violationCounts{} + var eventsSeen atomic.Int64 + + callbacks := &events.RepoStreamCallbacks{ + RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error { + n := eventsSeen.Add(1) + if cfg.didFilter == "" || cfg.didFilter == evt.Repo { + printCommit(evt, v, cfg) + } + if cfg.maxEvents > 0 && n >= cfg.maxEvents { + return errMaxEvents + } + return nil + }, + RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error { + eventsSeen.Add(1) + if cfg.didFilter == "" || cfg.didFilter == evt.Did { + printIdentity(evt) + } + return nil + }, + RepoAccount: func(evt *comatproto.SyncSubscribeRepos_Account) error { + eventsSeen.Add(1) + if cfg.didFilter == "" || cfg.didFilter == evt.Did { + printAccount(evt) + } + return nil + }, + RepoSync: func(evt *comatproto.SyncSubscribeRepos_Sync) error { + eventsSeen.Add(1) + if cfg.didFilter == "" || cfg.didFilter == evt.Did { + printSync(evt) + } + return nil + }, + RepoInfo: func(evt *comatproto.SyncSubscribeRepos_Info) error { + fmt.Printf("%s#info%s name=%s message=%s\n", cYellow, cReset, evt.Name, derefStr(evt.Message)) + return nil + }, + Error: func(evt *events.ErrorFrame) error { + fmt.Printf("%s#error%s %s: %s\n", cRed, cReset, evt.Error, evt.Message) + return nil + }, + } + + sched := sequential.NewScheduler("firehose-tap", callbacks.EventHandler) + streamErr := events.HandleRepoStream(ctx, conn, sched, nil) + if errors.Is(streamErr, errMaxEvents) { + streamErr = nil + } + + printSummary(eventsSeen.Load(), v, cfg) + + if cfg.validate && v.total() > 0 { + os.Exit(1) + } + return streamErr +} + +// errMaxEvents is returned from the commit handler to short-circuit +// HandleRepoStream once --max-events has been reached. +var errMaxEvents = errors.New("max events reached") + +func buildWSURL(endpoint string, cursor int64, cursorSet bool) (string, error) { + if !strings.Contains(endpoint, "://") { + endpoint = "wss://" + endpoint + } + u, err := url.Parse(endpoint) + if err != nil { + return "", fmt.Errorf("parse endpoint: %w", err) + } + switch u.Scheme { + case "wss", "ws": + // already a websocket URL + case "https": + u.Scheme = "wss" + case "http": + u.Scheme = "ws" + default: + return "", fmt.Errorf("unsupported scheme %q (use ws:// or wss://)", u.Scheme) + } + + if !strings.Contains(u.Path, "subscribeRepos") { + u.Path = strings.TrimRight(u.Path, "/") + "/xrpc/com.atproto.sync.subscribeRepos" + } + + if cursorSet { + q := u.Query() + q.Set("cursor", fmt.Sprintf("%d", cursor)) + u.RawQuery = q.Encode() + } + return u.String(), nil +} + +func printCommit(evt *comatproto.SyncSubscribeRepos_Commit, v *violationCounts, cfg *config) { + prevData := "(none)" + if evt.PrevData != nil { + prevData = shortCID(evt.PrevData.String()) + } + since := "(none)" + if evt.Since != nil { + since = *evt.Since + } + + // Spec: commits beyond the very first one should carry prevData. + if evt.Since != nil && evt.PrevData == nil { + v.missingPrevData.Add(1) + } + + fmt.Printf("%s#commit%s seq=%d repo=%s rev=%s since=%s prevData=%s ops=%d\n", + cGreen, cReset, evt.Seq, evt.Repo, evt.Rev, since, prevData, len(evt.Ops)) + + for _, op := range evt.Ops { + printOp(op, v, cfg) + } +} + +func printOp(op *comatproto.SyncSubscribeRepos_RepoOp, v *violationCounts, cfg *config) { + action := op.Action + + cidStr := "(nil)" + if op.Cid != nil { + cidStr = shortCID(op.Cid.String()) + } + prevStr := "(nil)" + if op.Prev != nil { + prevStr = shortCID(op.Prev.String()) + } + + violated := false + var noteParts []string + + switch action { + case "update", "delete": + if op.Prev == nil { + v.missingPrev.Add(1) + violated = true + noteParts = append(noteParts, "MISSING prev (Sync 1.1)") + } + case "create": + if op.Prev != nil { + v.createWithPrev.Add(1) + violated = true + noteParts = append(noteParts, "create MUST NOT have prev") + } + } + if action == "delete" && op.Cid != nil { + v.deleteWithCid.Add(1) + violated = true + noteParts = append(noteParts, "delete MUST have nil cid") + } + + // In non-verbose mode, only print ops that updated/deleted (the interesting + // Sync 1.1 cases) or that violated a rule. + if !cfg.verbose && action == "create" && !violated { + return + } + + actionColor := cCyan + if violated { + actionColor = cRed + } + note := "" + if len(noteParts) > 0 { + note = fmt.Sprintf(" %s[%s]%s", cRed, strings.Join(noteParts, "; "), cReset) + } + fmt.Printf(" %s%-6s%s %s cid=%s prev=%s%s\n", + actionColor, action, cReset, op.Path, cidStr, prevStr, note) +} + +func printIdentity(evt *comatproto.SyncSubscribeRepos_Identity) { + handle := "(nil)" + if evt.Handle != nil { + handle = *evt.Handle + } + fmt.Printf("%s#identity%s seq=%d did=%s handle=%s\n", + cYellow, cReset, evt.Seq, evt.Did, handle) +} + +func printAccount(evt *comatproto.SyncSubscribeRepos_Account) { + status := "(nil)" + if evt.Status != nil { + status = *evt.Status + } + fmt.Printf("%s#account%s seq=%d did=%s active=%v status=%s\n", + cYellow, cReset, evt.Seq, evt.Did, evt.Active, status) +} + +func printSync(evt *comatproto.SyncSubscribeRepos_Sync) { + fmt.Printf("%s#sync%s seq=%d did=%s rev=%s\n", + cYellow, cReset, evt.Seq, evt.Did, evt.Rev) +} + +func printSummary(seen int64, v *violationCounts, cfg *config) { + fmt.Fprintln(os.Stderr) + fmt.Fprintf(os.Stderr, "%s--- summary ---%s\n", cBold, cReset) + fmt.Fprintf(os.Stderr, "events seen: %d\n", seen) + if v.total() == 0 { + fmt.Fprintf(os.Stderr, "%sno Sync 1.1 violations%s\n", cGreen, cReset) + return + } + fmt.Fprintf(os.Stderr, "%sviolations:%s\n", cRed, cReset) + if n := v.missingPrev.Load(); n > 0 { + fmt.Fprintf(os.Stderr, " %d update/delete op(s) missing prev\n", n) + } + if n := v.createWithPrev.Load(); n > 0 { + fmt.Fprintf(os.Stderr, " %d create op(s) with non-nil prev\n", n) + } + if n := v.deleteWithCid.Load(); n > 0 { + fmt.Fprintf(os.Stderr, " %d delete op(s) with non-nil cid\n", n) + } + if n := v.missingPrevData.Load(); n > 0 { + fmt.Fprintf(os.Stderr, " %d commit(s) with since but missing prevData\n", n) + } + if !cfg.validate { + fmt.Fprintf(os.Stderr, "%s(--validate not set, exiting 0 anyway)%s\n", cDim, cReset) + } +} + +func shortCID(s string) string { + if len(s) <= 16 { + return s + } + return s[:8] + "…" + s[len(s)-6:] +} + +func derefStr(s *string) string { + if s == nil { + return "" + } + return *s +} diff --git a/config-appview.example.yaml b/config-appview.example.yaml index 17f8029..eb319bb 100644 --- a/config-appview.example.yaml +++ b/config-appview.example.yaml @@ -25,8 +25,6 @@ server: addr: :5000 # Public-facing URL for OAuth callbacks and JWT realm. Auto-detected if empty. base_url: "" - # DID of the hold service for blob storage, e.g. "did:web:hold01.atcr.io" (REQUIRED). - default_hold_did: "" # Allows HTTP (not HTTPS) for DID resolution and uses transition:generic OAuth scope. test_mode: false # Display name shown on OAuth authorization screens. @@ -34,10 +32,10 @@ server: # Short name used in page titles and browser tabs. client_short_name: ATCR # Separate domains for OCI registry API (e.g. ["buoy.cr"]). First is primary. Browser visits redirect to BaseURL. - registry_domains: [] - # DIDs of holds this appview manages billing for. Tier updates are pushed to these holds. + registry_domains: [127.0.0.1:5000, atcr.io] + # DIDs of holds this appview manages billing for (REQUIRED). The first entry is the default blob-storage hold. Tier updates are pushed to these holds. managed_holds: - - did:web:172.28.0.3%3A8080 + - did:web:172.28.0.3%3A8080 # Web UI settings. ui: # SQLite/libSQL database for OAuth sessions, stars, pull counts, and device approvals. @@ -164,5 +162,3 @@ billing: ai_advisor: true # Show supporter badge on user profiles for subscribers at this tier. supporter_badge: true - # Show supporter badge on hold owner profiles. - owner_badge: true diff --git a/deploy/upcloud/configs/appview.yaml.tmpl b/deploy/upcloud/configs/appview.yaml.tmpl index 56af52f..e48b4ce 100644 --- a/deploy/upcloud/configs/appview.yaml.tmpl +++ b/deploy/upcloud/configs/appview.yaml.tmpl @@ -10,7 +10,6 @@ log_shipper: server: addr: :5000 base_url: "https://seamark.dev" - default_hold_did: "{{.HoldDid}}" client_name: Seamark test_mode: false client_short_name: Seamark @@ -18,6 +17,8 @@ server: - "buoy.cr" - "bouy.cr" - "seamark.cr" + managed_holds: + - "{{.HoldDid}}" ui: database_path: "{{.BasePath}}/ui.db" theme: seamark diff --git a/deploy/upcloud/provision.go b/deploy/upcloud/provision.go index dbbb2bb..194bd76 100644 --- a/deploy/upcloud/provision.go +++ b/deploy/upcloud/provision.go @@ -367,10 +367,6 @@ func cmdProvision(token, zone, plan, sshKeyPath, s3Secret string, withScanner, w if appviewCreated || holdCreated { rootDir := projectRoot() - if err := runGenerate(rootDir); err != nil { - return fmt.Errorf("go generate: %w", err) - } - if err := runMakeBuildTrixie(rootDir); err != nil { return fmt.Errorf("build: %w", err) } @@ -422,9 +418,6 @@ func cmdProvision(token, zone, plan, sshKeyPath, s3Secret string, withScanner, w // not freshly created (the appviewCreated branch above already handled it). if state.LabelerEnabled && !appviewCreated { rootDir := projectRoot() - if err := runGenerate(rootDir); err != nil { - return fmt.Errorf("go generate: %w", err) - } labelerLocal := filepath.Join(rootDir, "bin", "atcr-labeler") if err := runMakeBuildTrixie(rootDir); err != nil { return fmt.Errorf("build labeler: %w", err) diff --git a/deploy/upcloud/update.go b/deploy/upcloud/update.go index 6b0c369..0da3d97 100644 --- a/deploy/upcloud/update.go +++ b/deploy/upcloud/update.go @@ -122,13 +122,10 @@ func cmdUpdate(target string, withScanner, withLabeler bool) error { return fmt.Errorf("unknown target: %s (use: all, appview, hold)", target) } - // Run go generate before building - if err := runGenerate(rootDir); err != nil { - return fmt.Errorf("go generate: %w", err) - } - // Build all binaries via `make build-trixie` so output links against // glibc 2.41 (the deploy target's glibc) regardless of the host's glibc. + // build-trixie depends on the `generate` make target, which runs + // `go generate ./...` on the host before invoking the trixie container. if err := runMakeBuildTrixie(rootDir); err != nil { return fmt.Errorf("build: %w", err) } @@ -347,17 +344,6 @@ func configValsFromState(state *InfraState) *ConfigValues { } } -// runGenerate runs go generate ./... in the given directory using host OS/arch -// (no cross-compilation env vars — generate tools must run on the build machine). -func runGenerate(dir string) error { - fmt.Println("Running go generate ./...") - cmd := exec.Command("go", "generate", "./...") - cmd.Dir = dir - cmd.Stdout = os.Stdout - cmd.Stderr = os.Stderr - return cmd.Run() -} - // runMakeBuildTrixie shells out to `make build-trixie`, which builds all // production binaries (appview, hold, credential-helper, labeler, scanner) // inside a Debian 13 container so they link against glibc 2.41. Centralizing diff --git a/docker-compose.yml b/docker-compose.yml index 43858e3..5be9d6c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,8 +17,9 @@ services: environment: # ATCR_SERVER_CLIENT_NAME: "Seamark" # ATCR_SERVER_CLIENT_SHORT_NAME: "Seamark" + # First entry is the default blob-storage hold. Comma-separate for multiple: + # ATCR_SERVER_MANAGED_HOLDS: "did:web:a,did:web:b" (Viper splits on commas). ATCR_SERVER_MANAGED_HOLDS: did:web:172.28.0.3%3A8080 - ATCR_SERVER_DEFAULT_HOLD_DID: did:web:172.28.0.3%3A8080 # Labeler URL (HTTP for dev — ParseLabelerURL accepts it directly so we don't # have to round-trip through did:web → https:// resolution). ATCR_LABELER_DID: did:web:172.28.0.4%3A5002 diff --git a/internal/testharness/harness.go b/internal/testharness/harness.go index 77285bc..a125aba 100644 --- a/internal/testharness/harness.go +++ b/internal/testharness/harness.go @@ -412,7 +412,7 @@ func buildAppViewConfig(addr, baseURL, holdDID, dbPath string) *appview.Config { cfg.LogLevel = "warn" cfg.Server.Addr = addr cfg.Server.BaseURL = baseURL - cfg.Server.DefaultHoldDID = holdDID + cfg.Server.ManagedHolds = []string{holdDID} cfg.Server.TestMode = true // Registry domain is a bare hostname (no port). DomainRoutingMiddleware // strips ports before matching, so "127.0.0.1" is what /v2/ requests diff --git a/lexicons/io/atcr/sailor/profile.json b/lexicons/io/atcr/sailor/profile.json index 8b6db6b..dc78eb5 100644 --- a/lexicons/io/atcr/sailor/profile.json +++ b/lexicons/io/atcr/sailor/profile.json @@ -19,6 +19,17 @@ "type": "boolean", "description": "Automatically delete manifest records that become untagged after a tag overwrite. Layers are cleaned up by hold garbage collection." }, + "ociClient": { + "type": "string", + "maxLength": 32, + "knownValues": ["docker", "podman", "buildah", "nerdctl", "crane", "none"], + "description": "Preferred client for pull commands (docker, podman, buildah, nerdctl, crane). 'none' shows the image reference only. Defaults to docker if empty." + }, + "registryDomain": { + "type": "string", + "maxLength": 255, + "description": "Preferred registry domain for UI display. Must be one of the appview's configured registry domains. Empty means the primary (first configured) domain." + }, "createdAt": { "type": "string", "format": "datetime", diff --git a/pkg/appview/config.go b/pkg/appview/config.go index ef97dce..6f3899a 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -47,9 +47,6 @@ type ServerConfig struct { // Public-facing URL for OAuth callbacks and JWT realm. BaseURL string `yaml:"base_url" comment:"Public-facing URL for OAuth callbacks and JWT realm. Auto-detected if empty."` - // DID of the default hold service for blob storage. - DefaultHoldDID string `yaml:"default_hold_did" comment:"DID of the hold service for blob storage, e.g. \"did:web:hold01.atcr.io\" (REQUIRED)."` - // Allows HTTP (not HTTPS) for DID resolution. TestMode bool `yaml:"test_mode" comment:"Allows HTTP (not HTTPS) for DID resolution and uses transition:generic OAuth scope."` @@ -62,8 +59,18 @@ type ServerConfig struct { // Separate domains for OCI registry API. First entry is the primary (used for JWT service name and UI display). RegistryDomains []string `yaml:"registry_domains" comment:"Separate domains for OCI registry API (e.g. [\"buoy.cr\"]). First is primary. Browser visits redirect to BaseURL."` - // DIDs of holds this appview manages billing for. - ManagedHolds []string `yaml:"managed_holds" comment:"DIDs of holds this appview manages billing for. Tier updates are pushed to these holds."` + // DIDs of holds this appview manages billing for. The first entry is also + // the default blob-storage hold (used when a user has no hold selected). + ManagedHolds []string `yaml:"managed_holds" comment:"DIDs of holds this appview manages billing for (REQUIRED). The first entry is the default blob-storage hold. Tier updates are pushed to these holds."` +} + +// PrimaryHoldDID returns the appview's default blob-storage hold, which is the +// first managed hold. Empty if none configured. +func (s ServerConfig) PrimaryHoldDID() string { + if len(s.ManagedHolds) > 0 { + return s.ManagedHolds[0] + } + return "" } // UIConfig defines web UI settings @@ -163,7 +170,6 @@ func setDefaults(v *viper.Viper) { // Server defaults v.SetDefault("server.addr", ":5000") v.SetDefault("server.base_url", "") - v.SetDefault("server.default_hold_did", "") v.SetDefault("server.test_mode", false) v.SetDefault("server.client_name", "AT Container Registry") v.SetDefault("server.client_short_name", "ATCR") @@ -235,7 +241,6 @@ func ExampleYAML() ([]byte, error) { cfg.Billing.Currency = "usd" cfg.Billing.SuccessURL = "{base_url}/settings/billing" cfg.Billing.CancelURL = "{base_url}/settings/billing" - cfg.Billing.OwnerBadge = true cfg.Billing.Tiers = []billing.BillingTierConfig{ {Name: "deckhand", Description: "Get started with basic storage", MaxWebhooks: 1}, {Name: "bosun", Description: "More storage with scan-on-push", StripePriceMonthly: "price_xxx", StripePriceYearly: "price_yyy", MaxWebhooks: 5, WebhookAllTriggers: true, SupporterBadge: true}, @@ -281,8 +286,8 @@ func LoadConfig(yamlPath string) (*Config, error) { } // Validation - if cfg.Server.DefaultHoldDID == "" { - return nil, fmt.Errorf("server.default_hold_did is required (env: ATCR_SERVER_DEFAULT_HOLD_DID)") + if len(cfg.Server.ManagedHolds) == 0 { + return nil, fmt.Errorf("server.managed_holds is required (at least one hold DID; env: ATCR_SERVER_MANAGED_HOLDS)") } if cfg.Labeler.DID != "" && !strings.HasPrefix(cfg.Labeler.DID, "did:") { return nil, fmt.Errorf("labeler.did must be a DID (did:plc:... or did:web:...), got %q", cfg.Labeler.DID) @@ -350,7 +355,7 @@ func buildDistributionConfig(cfg *Config, v *viper.Viper) (*configuration.Config distConfig.Storage = buildStorageConfig() // Middleware (ATProto resolver) - distConfig.Middleware = buildMiddlewareConfig(cfg.Server.DefaultHoldDID, cfg.Server.BaseURL, cfg.Server.TestMode) + distConfig.Middleware = buildMiddlewareConfig(cfg.Server.PrimaryHoldDID(), cfg.Server.BaseURL, cfg.Server.TestMode) // Auth (use values from cfg.Auth) // Realm always points to BaseURL where auth endpoints live diff --git a/pkg/appview/config_test.go b/pkg/appview/config_test.go index 8a34b1b..e38ea56 100644 --- a/pkg/appview/config_test.go +++ b/pkg/appview/config_test.go @@ -176,9 +176,9 @@ func TestLoadConfig(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { if tt.setHoldDID { - t.Setenv("ATCR_SERVER_DEFAULT_HOLD_DID", tt.envHoldDID) + t.Setenv("ATCR_SERVER_MANAGED_HOLDS", tt.envHoldDID) } else { - os.Unsetenv("ATCR_SERVER_DEFAULT_HOLD_DID") + os.Unsetenv("ATCR_SERVER_MANAGED_HOLDS") } // Clear other env vars to use defaults @@ -207,8 +207,8 @@ func TestLoadConfig(t *testing.T) { t.Errorf("HTTP addr = %v, want :5000", got.Server.Addr) } - if got.Server.DefaultHoldDID != tt.envHoldDID { - t.Errorf("default hold DID = %v, want %v", got.Server.DefaultHoldDID, tt.envHoldDID) + if got.Server.PrimaryHoldDID() != tt.envHoldDID { + t.Errorf("primary hold DID = %v, want %v", got.Server.PrimaryHoldDID(), tt.envHoldDID) } if got.UI.DatabasePath != "/var/lib/atcr/ui.db" { diff --git a/pkg/appview/db/cascade_delete_test.go b/pkg/appview/db/cascade_delete_test.go new file mode 100644 index 0000000..f5246be --- /dev/null +++ b/pkg/appview/db/cascade_delete_test.go @@ -0,0 +1,199 @@ +package db + +import ( + "database/sql" + "errors" + "testing" + "time" +) + +// seedCascadeFixture inserts a user and a single manifest. Returns the +// manifest's row id so callers can attach references (for the multi-arch case). +func seedCascadeFixture(t *testing.T, db *sql.DB, didStr, repo, digest string) int64 { + t.Helper() + + user := &User{ + DID: didStr, + Handle: "tester.example.com", + PDSEndpoint: "https://test.pds.example.com", + LastSeen: time.Now(), + } + if err := UpsertUser(db, user); err != nil { + t.Fatalf("UpsertUser: %v", err) + } + + id, err := InsertManifest(db, &Manifest{ + DID: didStr, + Repository: repo, + Digest: digest, + HoldEndpoint: "did:web:hold.example.com", + SchemaVersion: 2, + MediaType: "application/vnd.oci.image.manifest.v1+json", + CreatedAt: time.Now(), + }) + if err != nil { + t.Fatalf("InsertManifest: %v", err) + } + return id +} + +func TestGetTagDigest_ReturnsDigestForKnownTag(t *testing.T) { + db, err := InitDB(":memory:", LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + defer db.Close() + + const did = "did:plc:tagdigest" + const repo = "myapp" + const digest = "sha256:aaa" + + seedCascadeFixture(t, db, did, repo, digest) + + if err := UpsertTag(db, &Tag{ + DID: did, + Repository: repo, + Tag: "latest", + Digest: digest, + CreatedAt: time.Now(), + }); err != nil { + t.Fatalf("UpsertTag: %v", err) + } + + got, err := GetTagDigest(db, did, repo, "latest") + if err != nil { + t.Fatalf("GetTagDigest: %v", err) + } + if got != digest { + t.Errorf("digest mismatch: got %q want %q", got, digest) + } +} + +func TestGetTagDigest_UnknownTagReturnsErrNoRows(t *testing.T) { + db, err := InitDB(":memory:", LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + defer db.Close() + + seedCascadeFixture(t, db, "did:plc:tagdigest2", "myapp", "sha256:bbb") + + _, err = GetTagDigest(db, "did:plc:tagdigest2", "myapp", "does-not-exist") + if !errors.Is(err, sql.ErrNoRows) { + t.Errorf("expected sql.ErrNoRows for unknown tag, got %v", err) + } +} + +// TestShouldCascadeDeleteManifest_LastTagAndNoParent: the common case — +// digest has no remaining tags and is not referenced by any manifest list. +// Cascade should fire. +func TestShouldCascadeDeleteManifest_LastTagAndNoParent(t *testing.T) { + db, err := InitDB(":memory:", LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + defer db.Close() + + const did = "did:plc:cascade1" + const repo = "myapp" + const digest = "sha256:lonely" + + seedCascadeFixture(t, db, did, repo, digest) + + // No tags pointing to this digest, no manifest_references entries. + ok, err := ShouldCascadeDeleteManifest(db, did, repo, digest) + if err != nil { + t.Fatalf("ShouldCascadeDeleteManifest: %v", err) + } + if !ok { + t.Error("expected cascade=true when manifest is untagged and unreferenced") + } +} + +// TestShouldCascadeDeleteManifest_RemainingTagBlocks: another tag still +// points to this digest → keep the manifest alive. +func TestShouldCascadeDeleteManifest_RemainingTagBlocks(t *testing.T) { + db, err := InitDB(":memory:", LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + defer db.Close() + + const did = "did:plc:cascade2" + const repo = "myapp" + const digest = "sha256:shared" + + seedCascadeFixture(t, db, did, repo, digest) + + if err := UpsertTag(db, &Tag{ + DID: did, + Repository: repo, + Tag: "v1", + Digest: digest, + CreatedAt: time.Now(), + }); err != nil { + t.Fatalf("UpsertTag: %v", err) + } + + ok, err := ShouldCascadeDeleteManifest(db, did, repo, digest) + if err != nil { + t.Fatalf("ShouldCascadeDeleteManifest: %v", err) + } + if ok { + t.Error("expected cascade=false when another tag still points to the digest") + } +} + +// TestShouldCascadeDeleteManifest_MultiArchChildBlocks: the digest is a child +// of a manifest list (multi-arch parent). Even with no tags, deleting it +// would orphan the parent's reference, so we must NOT cascade. +func TestShouldCascadeDeleteManifest_MultiArchChildBlocks(t *testing.T) { + db, err := InitDB(":memory:", LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + defer db.Close() + + const did = "did:plc:cascade3" + const repo = "myapp" + const childDigest = "sha256:amd64child" + const parentDigest = "sha256:multiarchparent" + + // Insert the child manifest fixture. + seedCascadeFixture(t, db, did, repo, childDigest) + + // Insert a separate parent (manifest list) and attach a manifest_reference + // from parent → child. + parentID, err := InsertManifest(db, &Manifest{ + DID: did, + Repository: repo, + Digest: parentDigest, + HoldEndpoint: "did:web:hold.example.com", + SchemaVersion: 2, + MediaType: "application/vnd.oci.image.index.v1+json", + CreatedAt: time.Now(), + }) + if err != nil { + t.Fatalf("InsertManifest(parent): %v", err) + } + + if err := InsertManifestReference(db, &ManifestReference{ + ManifestID: parentID, + Digest: childDigest, + Size: 1234, + MediaType: "application/vnd.oci.image.manifest.v1+json", + PlatformArchitecture: "amd64", + PlatformOS: "linux", + ReferenceIndex: 0, + }); err != nil { + t.Fatalf("InsertManifestReference: %v", err) + } + + ok, err := ShouldCascadeDeleteManifest(db, did, repo, childDigest) + if err != nil { + t.Fatalf("ShouldCascadeDeleteManifest: %v", err) + } + if ok { + t.Error("expected cascade=false when digest is a child of a manifest list") + } +} diff --git a/pkg/appview/db/migrations/0027_add_registry_domain.yaml b/pkg/appview/db/migrations/0027_add_registry_domain.yaml new file mode 100644 index 0000000..da3d01e --- /dev/null +++ b/pkg/appview/db/migrations/0027_add_registry_domain.yaml @@ -0,0 +1,3 @@ +description: Add registry_domain column to users table for preferred registry domain +query: | + ALTER TABLE users ADD COLUMN registry_domain TEXT DEFAULT ''; diff --git a/pkg/appview/db/models.go b/pkg/appview/db/models.go index 0a31944..ef5f88a 100644 --- a/pkg/appview/db/models.go +++ b/pkg/appview/db/models.go @@ -10,6 +10,7 @@ type User struct { Avatar string DefaultHoldDID string OciClient string + RegistryDomain string LastSeen time.Time } diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 1e1eaff..c54d0eb 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -442,12 +442,12 @@ func GetRepositoryMetadata(db DBTX, did string, repository string) (map[string]s // GetUserByDID retrieves a user by DID func GetUserByDID(db DBTX, did string) (*User, error) { var user User - var avatar, defaultHoldDID, ociClient sql.NullString + var avatar, defaultHoldDID, ociClient, registryDomain sql.NullString err := db.QueryRow(` - SELECT did, handle, pds_endpoint, avatar, default_hold_did, oci_client, last_seen + SELECT did, handle, pds_endpoint, avatar, default_hold_did, oci_client, registry_domain, last_seen FROM users WHERE did = ? - `, did).Scan(&user.DID, &user.Handle, &user.PDSEndpoint, &avatar, &defaultHoldDID, &ociClient, &user.LastSeen) + `, did).Scan(&user.DID, &user.Handle, &user.PDSEndpoint, &avatar, &defaultHoldDID, &ociClient, ®istryDomain, &user.LastSeen) if err == sql.ErrNoRows { return nil, nil @@ -465,6 +465,9 @@ func GetUserByDID(db DBTX, did string) (*User, error) { if ociClient.Valid { user.OciClient = ociClient.String } + if registryDomain.Valid { + user.RegistryDomain = registryDomain.String + } return &user, nil } @@ -472,12 +475,12 @@ func GetUserByDID(db DBTX, did string) (*User, error) { // GetUserByHandle retrieves a user by handle func GetUserByHandle(db DBTX, handle string) (*User, error) { var user User - var avatar, defaultHoldDID, ociClient sql.NullString + var avatar, defaultHoldDID, ociClient, registryDomain sql.NullString err := db.QueryRow(` - SELECT did, handle, pds_endpoint, avatar, default_hold_did, oci_client, last_seen + SELECT did, handle, pds_endpoint, avatar, default_hold_did, oci_client, registry_domain, last_seen FROM users WHERE handle = ? - `, handle).Scan(&user.DID, &user.Handle, &user.PDSEndpoint, &avatar, &defaultHoldDID, &ociClient, &user.LastSeen) + `, handle).Scan(&user.DID, &user.Handle, &user.PDSEndpoint, &avatar, &defaultHoldDID, &ociClient, ®istryDomain, &user.LastSeen) if err == sql.ErrNoRows { return nil, nil @@ -495,6 +498,9 @@ func GetUserByHandle(db DBTX, handle string) (*User, error) { if ociClient.Valid { user.OciClient = ociClient.String } + if registryDomain.Valid { + user.RegistryDomain = registryDomain.String + } return &user, nil } @@ -591,6 +597,15 @@ func UpdateUserOciClient(db DBTX, did string, ociClient string) error { return err } +// UpdateUserRegistryDomain updates a user's cached preferred registry domain. +// An empty string means "use the primary (first configured) domain". +func UpdateUserRegistryDomain(db DBTX, did string, registryDomain string) error { + _, err := db.Exec(` + UPDATE users SET registry_domain = ? WHERE did = ? + `, registryDomain, did) + return err +} + // GetUserHoldDID returns the hold DID for a user. Uses cached default_hold_did // if available, otherwise falls back to the most recent manifest's hold_endpoint. func GetUserHoldDID(db DBTX, did string) string { @@ -1541,6 +1556,37 @@ func IsManifestReferenced(db DBTX, did, digest string) (bool, error) { return count > 0, nil } +// GetTagDigest returns the manifest digest that a tag points to. +// Returns sql.ErrNoRows if the tag does not exist. +func GetTagDigest(db DBTX, did, repository, tag string) (string, error) { + var digest string + err := db.QueryRow(` + SELECT digest FROM tags + WHERE did = ? AND repository = ? AND tag = ? + `, did, repository, tag).Scan(&digest) + return digest, err +} + +// ShouldCascadeDeleteManifest returns true iff a manifest can be safely +// deleted after a tag is removed: it has no remaining tags AND is not +// referenced by any manifest list (multi-arch parent). Manifest-list +// children must be preserved even when untagged, since their parent index +// still depends on them. +func ShouldCascadeDeleteManifest(db DBTX, did, repository, digest string) (bool, error) { + tagged, err := IsManifestTagged(db, did, repository, digest) + if err != nil { + return false, err + } + if tagged { + return false, nil + } + referenced, err := IsManifestReferenced(db, did, digest) + if err != nil { + return false, err + } + return !referenced, nil +} + // IsManifestTagged checks if a manifest has any tags func IsManifestTagged(db DBTX, did, repository, digest string) (bool, error) { var count int diff --git a/pkg/appview/db/queries_test.go b/pkg/appview/db/queries_test.go index 31510e3..ba19fe7 100644 --- a/pkg/appview/db/queries_test.go +++ b/pkg/appview/db/queries_test.go @@ -987,6 +987,64 @@ func TestGetTagsWithPlatforms(t *testing.T) { _ = manifestID1 } +func TestUserRegistryDomain(t *testing.T) { + db, err := InitDB(":memory:", LibsqlConfig{}) + if err != nil { + t.Fatalf("Failed to init database: %v", err) + } + defer db.Close() + + user := &User{ + DID: "did:plc:domainpref", + Handle: "pref.bsky.social", + PDSEndpoint: "https://bsky.social", + LastSeen: time.Now(), + } + if err := UpsertUser(db, user); err != nil { + t.Fatalf("Failed to upsert user: %v", err) + } + + // Default is empty string (NULL-safe scan). + got, err := GetUserByDID(db, user.DID) + if err != nil { + t.Fatalf("GetUserByDID failed: %v", err) + } + if got.RegistryDomain != "" { + t.Errorf("Expected empty registry domain by default, got %q", got.RegistryDomain) + } + + // Update and confirm it round-trips through both lookups. + if err := UpdateUserRegistryDomain(db, user.DID, "buoy.cr"); err != nil { + t.Fatalf("UpdateUserRegistryDomain failed: %v", err) + } + byDID, err := GetUserByDID(db, user.DID) + if err != nil { + t.Fatalf("GetUserByDID failed: %v", err) + } + if byDID.RegistryDomain != "buoy.cr" { + t.Errorf("GetUserByDID: expected 'buoy.cr', got %q", byDID.RegistryDomain) + } + byHandle, err := GetUserByHandle(db, user.Handle) + if err != nil { + t.Fatalf("GetUserByHandle failed: %v", err) + } + if byHandle.RegistryDomain != "buoy.cr" { + t.Errorf("GetUserByHandle: expected 'buoy.cr', got %q", byHandle.RegistryDomain) + } + + // Clearing the preference propagates back to empty. + if err := UpdateUserRegistryDomain(db, user.DID, ""); err != nil { + t.Fatalf("UpdateUserRegistryDomain (clear) failed: %v", err) + } + cleared, err := GetUserByDID(db, user.DID) + if err != nil { + t.Fatalf("GetUserByDID failed: %v", err) + } + if cleared.RegistryDomain != "" { + t.Errorf("Expected empty registry domain after clear, got %q", cleared.RegistryDomain) + } +} + func TestUpdateUserHandle(t *testing.T) { // Create in-memory test database db, err := InitDB(":memory:", LibsqlConfig{}) diff --git a/pkg/appview/db/schema.sql b/pkg/appview/db/schema.sql index 719f72d..1563318 100644 --- a/pkg/appview/db/schema.sql +++ b/pkg/appview/db/schema.sql @@ -14,6 +14,7 @@ CREATE TABLE IF NOT EXISTS users ( avatar TEXT, default_hold_did TEXT, oci_client TEXT DEFAULT '', + registry_domain TEXT DEFAULT '', last_seen TIMESTAMP NOT NULL, UNIQUE(handle) ); diff --git a/pkg/appview/handlers/base.go b/pkg/appview/handlers/base.go index 3276128..34650d1 100644 --- a/pkg/appview/handlers/base.go +++ b/pkg/appview/handlers/base.go @@ -19,9 +19,10 @@ import ( // Route registration becomes simply &Handler{base} for everything. type BaseUIHandler struct { // Display - Templates *template.Template - RegistryURL string // Docker registry domain (e.g., "buoy.cr" or "atcr.io") - SiteURL string // Website domain (e.g., "seamark.dev" or "atcr.io") + Templates *template.Template + RegistryURL string // Primary Docker registry domain (e.g., "buoy.cr" or "atcr.io") + RegistryDomains []string // All configured registry domains; users may pick one as their default + SiteURL string // Website domain (e.g., "seamark.dev" or "atcr.io") // Database (handlers choose which to use) DB *sql.DB // Write access diff --git a/pkg/appview/handlers/common.go b/pkg/appview/handlers/common.go index 4a11b38..d1bc30d 100644 --- a/pkg/appview/handlers/common.go +++ b/pkg/appview/handlers/common.go @@ -2,6 +2,7 @@ package handlers import ( "net/http" + "slices" "strings" "atcr.io/pkg/appview/db" @@ -22,17 +23,28 @@ type PageData struct { CurrentPath string // Request path (used for OAuth return_to) } +// resolveRegistryURL returns the user's preferred registry domain when it is +// non-empty and still one of the configured domains; otherwise the primary. +// This keeps a stale or removed preference from breaking the displayed commands. +func resolveRegistryURL(primary string, domains []string, pref string) string { + if pref != "" && slices.Contains(domains, pref) { + return pref + } + return primary +} + // NewPageData creates a PageData struct with common fields populated from the request func NewPageData(r *http.Request, h *BaseUIHandler) PageData { user := middleware.GetUser(r) - var ociClient string + var ociClient, registryPref string if user != nil { ociClient = user.OciClient + registryPref = user.RegistryDomain } return PageData{ User: user, Query: r.URL.Query().Get("q"), - RegistryURL: h.RegistryURL, + RegistryURL: resolveRegistryURL(h.RegistryURL, h.RegistryDomains, registryPref), SiteURL: h.SiteURL, ClientName: h.ClientName, ClientShortName: h.ClientShortName, diff --git a/pkg/appview/handlers/common_test.go b/pkg/appview/handlers/common_test.go index b062de9..3c9129a 100644 --- a/pkg/appview/handlers/common_test.go +++ b/pkg/appview/handlers/common_test.go @@ -2,6 +2,34 @@ package handlers import "testing" +func TestResolveRegistryURL(t *testing.T) { + primary := "buoy.cr" + domains := []string{"buoy.cr", "atcr.io"} + + tests := []struct { + name string + domains []string + pref string + want string + }{ + {"empty pref falls back to primary", domains, "", "buoy.cr"}, + {"valid pref is used", domains, "atcr.io", "atcr.io"}, + {"pref equal to primary", domains, "buoy.cr", "buoy.cr"}, + {"stale pref not in list falls back", domains, "gone.example", "buoy.cr"}, + {"nil domains falls back", nil, "atcr.io", "buoy.cr"}, + {"empty domains falls back", []string{}, "atcr.io", "buoy.cr"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveRegistryURL(primary, tt.domains, tt.pref) + if got != tt.want { + t.Errorf("resolveRegistryURL(%q, %v, %q) = %q, want %q", primary, tt.domains, tt.pref, got, tt.want) + } + }) + } +} + func TestTrimRegistryURL(t *testing.T) { tests := []struct { name string diff --git a/pkg/appview/handlers/images.go b/pkg/appview/handlers/images.go index d404afa..7b6f1a5 100644 --- a/pkg/appview/handlers/images.go +++ b/pkg/appview/handlers/images.go @@ -1,10 +1,12 @@ package handlers import ( + "database/sql" "encoding/json" "errors" "fmt" "io" + "log/slog" "net/http" "strings" "time" @@ -42,15 +44,30 @@ func (h *DeleteTagHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { repo := req.Repo tag := req.Tag + // Look up the digest the tag points to (and the hold that owns its blobs) + // before we delete anything — we need both for the cascade decision and + // for purging the manifest's blobs on the hold. Missing-tag means there's + // nothing to delete; reuse 404 semantics. + digest, err := db.GetTagDigest(h.ReadOnlyDB, user.DID, repo, tag) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.Error(w, "Tag not found", http.StatusNotFound) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + var holdDID string + if cached, err := db.GetManifestDetail(h.ReadOnlyDB, user.DID, repo, digest); err == nil && cached != nil { + holdDID = cached.HoldEndpoint + } + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) - // Compute rkey for tag record (repository_tag with slashes replaced) - rkey := fmt.Sprintf("%s_%s", repo, tag) - rkey = strings.ReplaceAll(rkey, "/", "-") - - // Delete from PDS first - if err := pdsClient.DeleteRecord(r.Context(), atproto.TagCollection, rkey); err != nil { + // Delete the tag record from PDS first. + tagRKey := atproto.RepositoryTagToRKey(repo, tag) + if err := pdsClient.DeleteRecord(r.Context(), atproto.TagCollection, tagRKey); err != nil { // Check if OAuth error - if so, invalidate sessions and return 401 if handleOAuthError(r.Context(), h.Refresher, user.DID, err) { http.Error(w, "Authentication failed, please log in again", http.StatusUnauthorized) @@ -66,7 +83,32 @@ func (h *DeleteTagHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // Return empty response (HTMX will swap out the element) + // If this was the last tag pointing to the digest and the digest isn't a + // child of any manifest list, cascade-delete the manifest too — that's + // what the user almost always wants when they click "delete tag" on an + // only-tagged image. Failures here are non-fatal: the tag is already + // gone, so the worst case is a leftover untagged manifest the user can + // clean up via "Delete untagged". + shouldCascade, err := db.ShouldCascadeDeleteManifest(h.ReadOnlyDB, user.DID, repo, digest) + if err != nil { + slog.Warn("delete-tag: failed to evaluate cascade", "did", user.DID, "repo", repo, "digest", digest, "error", err) + } else if shouldCascade { + manifestRKey := strings.TrimPrefix(digest, "sha256:") + if err := pdsClient.DeleteRecord(r.Context(), atproto.ManifestCollection, manifestRKey); err != nil { + slog.Warn("delete-tag: cascade PDS delete failed", "did", user.DID, "digest", digest, "error", err) + } else { + if err := db.DeleteManifest(h.DB, user.DID, repo, digest); err != nil { + slog.Warn("delete-tag: cascade DB delete failed", "did", user.DID, "digest", digest, "error", err) + } + purgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, atproto.BuildManifestURI(user.DID, digest)) + } + } + + // Return empty response (HTMX will swap out the element). The clicked + // row was tag-specific, so removing it is correct whether or not the + // underlying manifest was also cascaded. In the "multi-arch parent + // preserved the manifest" case, the untagged manifest will reappear as + // its own row on the next page load — which matches reality. w.WriteHeader(http.StatusOK) } diff --git a/pkg/appview/handlers/settings.go b/pkg/appview/handlers/settings.go index 9798f98..302a9af 100644 --- a/pkg/appview/handlers/settings.go +++ b/pkg/appview/handlers/settings.go @@ -8,6 +8,7 @@ import ( "log/slog" "net/http" "net/url" + "slices" "strings" "time" @@ -77,6 +78,7 @@ type settingsProfile struct { DefaultHold string AutoRemoveUntagged bool OciClient string + RegistryDomain string AIAdvisorEnabled bool HasAIAdvisorAccess bool } @@ -87,16 +89,17 @@ type settingsProfile struct { // hold_selector template from doing filter-the-same-list-twice gymnastics. type settingsPageData struct { PageData - Meta *PageMeta - ActiveTab string - Tabs []settingsTab - Profile settingsProfile - ActiveHold *HoldDisplay - OtherHolds []HoldDisplay - MemberHolds []HoldDisplay - EligibleHolds []HoldDisplay - WebhooksData webhooksTemplateData - Subscription SubscriptionDisplay + Meta *PageMeta + ActiveTab string + Tabs []settingsTab + Profile settingsProfile + RegistryDomains []string + ActiveHold *HoldDisplay + OtherHolds []HoldDisplay + MemberHolds []HoldDisplay + EligibleHolds []HoldDisplay + WebhooksData webhooksTemplateData + Subscription SubscriptionDisplay } // ServeHTTP redirects /settings to /settings/user. @@ -151,8 +154,10 @@ func (h *SettingsHandler) ServeTab(tab string) http.HandlerFunc { DefaultHold: profile.DefaultHold, AutoRemoveUntagged: profile.AutoRemoveUntagged, OciClient: profile.OciClient, + RegistryDomain: profile.RegistryDomain, AIAdvisorEnabled: profile.AIAdvisorEnabled == nil || *profile.AIAdvisorEnabled, }, + RegistryDomains: h.RegistryDomains, } if h.BillingManager != nil { data.Profile.HasAIAdvisorAccess = h.BillingManager.HasAIAdvisor(user.DID) @@ -564,6 +569,58 @@ func (h *UpdateOciClientHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques w.WriteHeader(http.StatusNoContent) } +// UpdateRegistryDomainHandler handles updating the preferred registry domain +type UpdateRegistryDomainHandler struct { + BaseUIHandler +} + +func (h *UpdateRegistryDomainHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + user := middleware.GetUser(r) + if user == nil { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + registryDomain := r.FormValue("registry_domain") + // Empty means "use the primary domain". Any non-empty value must be one of + // the configured registry domains. + if registryDomain != "" && !slices.Contains(h.RegistryDomains, registryDomain) { + http.Error(w, "Invalid registry domain", http.StatusBadRequest) + return + } + + // Store empty string when the primary (first configured) domain is selected, + // so the preference tracks the primary even if the admin reorders domains. + if len(h.RegistryDomains) > 0 && registryDomain == h.RegistryDomains[0] { + registryDomain = "" + } + + // Create ATProto client with session provider + client := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) + + // Fetch existing profile + profile, err := storage.GetProfile(r.Context(), client) + if err != nil || profile == nil { + http.Error(w, "Failed to fetch profile", http.StatusInternalServerError) + return + } + + profile.RegistryDomain = registryDomain + profile.UpdatedAt = time.Now() + + if err := storage.UpdateProfile(r.Context(), client, profile); err != nil { + http.Error(w, "Failed to update profile: "+err.Error(), http.StatusInternalServerError) + return + } + + // Cache locally + if h.DB != nil { + _ = db.UpdateUserRegistryDomain(h.DB, user.DID, registryDomain) + } + + w.WriteHeader(http.StatusNoContent) +} + // UpdateAIAdvisorHandler handles toggling the AI Image Advisor setting type UpdateAIAdvisorHandler struct { BaseUIHandler diff --git a/pkg/appview/jetstream/processor.go b/pkg/appview/jetstream/processor.go index 6032663..4d8d522 100644 --- a/pkg/appview/jetstream/processor.go +++ b/pkg/appview/jetstream/processor.go @@ -542,6 +542,11 @@ func (p *Processor) ProcessSailorProfile(ctx context.Context, did string, record } } + // Cache preferred registry domain (write unconditionally so clearing it propagates) + if err := db.UpdateUserRegistryDomain(p.db, did, profileRecord.RegistryDomain); err != nil { + slog.Warn("Failed to cache registry domain preference", "component", "processor", "did", did, "registryDomain", profileRecord.RegistryDomain, "error", err) + } + // Skip hold processing if no default hold set if profileRecord.DefaultHold == "" { return nil diff --git a/pkg/appview/jetstream/processor_test.go b/pkg/appview/jetstream/processor_test.go index 93a478c..a24c451 100644 --- a/pkg/appview/jetstream/processor_test.go +++ b/pkg/appview/jetstream/processor_test.go @@ -43,6 +43,7 @@ func setupTestDB(t *testing.T) *sql.DB { avatar TEXT, default_hold_did TEXT, oci_client TEXT DEFAULT '', + registry_domain TEXT DEFAULT '', last_seen TIMESTAMP NOT NULL ); @@ -533,6 +534,46 @@ func TestProcessStar_InvalidRecord(t *testing.T) { } } +func TestProcessSailorProfile(t *testing.T) { + database := setupTestDB(t) + defer database.Close() + + if _, err := database.Exec( + `INSERT INTO users (did, handle, pds_endpoint, last_seen) VALUES (?, ?, ?, ?)`, + "did:plc:profile123", "profile.test", "https://pds.example.com", time.Now()); err != nil { + t.Fatalf("Failed to insert user: %v", err) + } + + p := NewProcessor(database, false, nil) + ctx := context.Background() + + readDomain := func() string { + var d string + if err := database.QueryRow("SELECT registry_domain FROM users WHERE did = ?", "did:plc:profile123").Scan(&d); err != nil { + t.Fatalf("query registry_domain: %v", err) + } + return d + } + + // A profile with registryDomain set caches it on the user row. + rec := []byte(`{"$type":"io.atcr.sailor.profile","registryDomain":"buoy.cr","createdAt":"2025-01-01T00:00:00Z"}`) + if err := p.ProcessSailorProfile(ctx, "did:plc:profile123", rec, nil); err != nil { + t.Fatalf("ProcessSailorProfile failed: %v", err) + } + if got := readDomain(); got != "buoy.cr" { + t.Errorf("expected cached registry domain 'buoy.cr', got %q", got) + } + + // A profile with an empty registryDomain clears the cached value. + recEmpty := []byte(`{"$type":"io.atcr.sailor.profile","createdAt":"2025-01-01T00:00:00Z"}`) + if err := p.ProcessSailorProfile(ctx, "did:plc:profile123", recEmpty, nil); err != nil { + t.Fatalf("ProcessSailorProfile (empty) failed: %v", err) + } + if got := readDomain(); got != "" { + t.Errorf("expected cleared registry domain, got %q", got) + } +} + func TestProcessManifest_Duplicate(t *testing.T) { database := setupTestDB(t) defer database.Close() diff --git a/pkg/appview/routes/routes.go b/pkg/appview/routes/routes.go index e3af3dc..2859509 100644 --- a/pkg/appview/routes/routes.go +++ b/pkg/appview/routes/routes.go @@ -33,7 +33,8 @@ type UIDependencies struct { OAuthStore *db.OAuthStore Refresher *oauth.Refresher BaseURL string - RegistryDomain string // Separate OCI registry domain (e.g., "buoy.cr"); empty = same as BaseURL + RegistryDomain string // Separate OCI registry domain (e.g., "buoy.cr"); empty = same as BaseURL + RegistryDomains []string // All configured registry domains; users may pick one as their default DeviceStore *db.DeviceStore HealthChecker *holdhealth.Checker ReadmeFetcher *readme.Fetcher @@ -63,6 +64,7 @@ func RegisterUIRoutes(router chi.Router, deps UIDependencies) { base := uihandlers.BaseUIHandler{ Templates: deps.Templates, RegistryURL: registryURL, + RegistryDomains: deps.RegistryDomains, SiteURL: siteURL, DB: deps.Database, ReadOnlyDB: deps.ReadOnlyDB, @@ -208,6 +210,7 @@ func RegisterUIRoutes(router chi.Router, deps UIDependencies) { r.Post("/api/profile/default-hold", (&uihandlers.UpdateDefaultHoldHandler{BaseUIHandler: base}).ServeHTTP) r.Post("/api/profile/auto-remove-untagged", (&uihandlers.UpdateAutoRemoveUntaggedHandler{BaseUIHandler: base}).ServeHTTP) r.Post("/api/profile/oci-client", (&uihandlers.UpdateOciClientHandler{BaseUIHandler: base}).ServeHTTP) + r.Post("/api/profile/registry-domain", (&uihandlers.UpdateRegistryDomainHandler{BaseUIHandler: base}).ServeHTTP) r.Post("/api/profile/ai-advisor", (&uihandlers.UpdateAIAdvisorHandler{BaseUIHandler: base}).ServeHTTP) // Subscription management diff --git a/pkg/appview/server.go b/pkg/appview/server.go index c077573..8582069 100644 --- a/pkg/appview/server.go +++ b/pkg/appview/server.go @@ -194,7 +194,7 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, slog.Info("Using SQLite for device storage") baseURL := cfg.Server.BaseURL - defaultHoldDID := cfg.Server.DefaultHoldDID + defaultHoldDID := cfg.Server.PrimaryHoldDID() testMode := cfg.Server.TestMode slog.Debug("Base URL for OAuth", "base_url", baseURL) @@ -340,6 +340,7 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, Refresher: s.Refresher, BaseURL: baseURL, RegistryDomain: primaryRegistryDomain(cfg.Server.RegistryDomains), + RegistryDomains: cfg.Server.RegistryDomains, DeviceStore: s.DeviceStore, HealthChecker: s.HealthChecker, ReadmeFetcher: s.ReadmeFetcher, @@ -909,7 +910,7 @@ func (s *AppViewServer) initializeJetstream() { if s.Config.Jetstream.BackfillEnabled { relayEndpoints := s.Config.Jetstream.RelayEndpoints - defaultHoldDID := s.Config.Server.DefaultHoldDID + defaultHoldDID := s.Config.Server.PrimaryHoldDID() testMode := s.Config.Server.TestMode backfillWorker, err := jetstream.NewBackfillWorker(s.Database, relayEndpoints, defaultHoldDID, testMode, s.Refresher) diff --git a/pkg/appview/templates/pages/user.html b/pkg/appview/templates/pages/user.html index 858fd9a..d54a014 100644 --- a/pkg/appview/templates/pages/user.html +++ b/pkg/appview/templates/pages/user.html @@ -33,7 +33,7 @@ {{ end }}

{{ .ViewedUser.Handle }}

- {{ if or (eq .SupporterBadge "Captain") (eq .SupporterBadge "owner") }} + {{ if eq .SupporterBadge "Captain" }} {{ .SupporterBadge }} {{ else if .SupporterBadge }} {{ .SupporterBadge }} diff --git a/pkg/appview/templates/partials/settings-panel-user.html b/pkg/appview/templates/partials/settings-panel-user.html index fcc1cf1..75f2c0c 100644 --- a/pkg/appview/templates/partials/settings-panel-user.html +++ b/pkg/appview/templates/partials/settings-panel-user.html @@ -5,13 +5,13 @@

Customize your experience across the site.

-
-
+
+

Sets the pull command shown on repository pages. Choose Image reference only to copy without a command prefix.

{{ $oci := .Profile.OciClient }} - + {{ range $i, $d := .RegistryDomains }} + + {{ end }} + +
+ {{ end }} + {{ if .AIAdvisorEnabled }}
@@ -44,8 +63,8 @@
AI Image Advisor

Analyze your container images for optimization suggestions using AI.

-

- Upgrade your plan to enable this feature. +

+ Upgrade your plan to enable this feature.

{{ end }} diff --git a/pkg/atproto/lexicon.go b/pkg/atproto/lexicon.go index ef918da..2e3efd3 100644 --- a/pkg/atproto/lexicon.go +++ b/pkg/atproto/lexicon.go @@ -356,6 +356,10 @@ type SailorProfileRecord struct { // "none" means image reference only (no ` pull ` prefix). Defaults to "docker" if empty. OciClient string `json:"ociClient,omitempty"` + // RegistryDomain is the user's preferred registry domain for UI display. + // Must be one of the appview's configured registry_domains. Empty = primary (first configured). + RegistryDomain string `json:"registryDomain,omitempty"` + // AIAdvisorEnabled controls whether the AI Image Advisor feature is active for this user. // nil = default (enabled if user has billing access), false = explicitly disabled. AIAdvisorEnabled *bool `json:"aiAdvisorEnabled,omitempty"` diff --git a/pkg/billing/config.go b/pkg/billing/config.go index 59652a6..4e917a0 100644 --- a/pkg/billing/config.go +++ b/pkg/billing/config.go @@ -23,9 +23,6 @@ type Config struct { // Subscription tiers with Stripe price IDs. Tiers []BillingTierConfig `yaml:"tiers" comment:"Subscription tiers ordered by rank (lowest to highest)."` - - // Whether hold owners get a supporter badge on their profile. - OwnerBadge bool `yaml:"owner_badge" comment:"Show supporter badge on hold owner profiles."` } // BillingTierConfig represents a single tier with optional Stripe pricing. diff --git a/pkg/hold/admin/admin.go b/pkg/hold/admin/admin.go index 3c02921..e00f7c6 100644 --- a/pkg/hold/admin/admin.go +++ b/pkg/hold/admin/admin.go @@ -92,20 +92,10 @@ type AdminUI struct { sessions map[string]*AdminSession sessionsMu sync.RWMutex - // scan-backfill state — runs as a background goroutine on click; the - // status endpoint reads this for progress polling. Only one run at a - // time (idempotent, so re-running is safe but pointless). - scanBackfill scanBackfillState -} - -// scanBackfillState tracks the in-flight scan-status backfill run. -type scanBackfillState struct { - mu sync.Mutex - running bool - startedAt time.Time - current *pds.ScanBackfillResult // running totals (snapshot) - result *pds.ScanBackfillResult // final result, set when running=false - err string // last error (running ends with err set) + // jobs tracks long-running background admin operations (bulk crew tier + // remap, crew import, scan-record backfill). The kickoff handler returns a + // progress fragment that polls /admin/api/jobs/{key}/status. See jobs.go. + jobs jobRegistry } // adminContextKey is used to store session data in request context @@ -544,11 +534,15 @@ func (ui *AdminUI) RegisterRoutes(r chi.Router) { r.Get("/admin/api/relay/status", ui.handleRelayStatus) r.Get("/admin/api/crew/member", ui.handleCrewMemberInfo) - // Scan-record backfill: kicks off a background run and returns a - // progress fragment that polls /status. Use Accept:application/json - // for a synchronous JSON response (curl-friendly). + // Scan-record backfill: kicks off a background job and returns a + // progress fragment that polls /admin/api/jobs/scan-backfill/status. + // Use Accept:application/json for a synchronous JSON response. r.Post("/admin/api/scan-backfill", ui.handleScanBackfill) - r.Get("/admin/api/scan-backfill/status", ui.handleScanBackfillStatus) + + // Generic background-job status, polled by progress fragments. Serves + // every job registered via startJob (see jobs.go) — crew tier remap, + // crew import, scan-record backfill. + r.Get("/admin/api/jobs/{key}/status", ui.handleJobStatus) // Logout r.Post("/admin/auth/logout", ui.handleLogout) diff --git a/pkg/hold/admin/handlers_crew.go b/pkg/hold/admin/handlers_crew.go index 41bb1e4..012ec0c 100644 --- a/pkg/hold/admin/handlers_crew.go +++ b/pkg/hold/admin/handlers_crew.go @@ -462,11 +462,16 @@ func (ui *AdminUI) handleCrewUpdate(w http.ResponseWriter, r *http.Request) { // because `to` is still validated against the live tier list. func (ui *AdminUI) handleCrewRemapTier(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - defer clearFlash(w) + + // This endpoint is driven by htmx (hx-post on the reconciliation card), so + // synchronous validation failures must render a 200 fragment, not a 302 — + // an htmx swap can't follow a redirect cleanly. + renderErr := func(msg string) { + ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{msg}) + } if err := r.ParseForm(); err != nil { - setFlash(w, r, "error", "Invalid form data") - http.Redirect(w, r, "/admin#crew", http.StatusFound) + renderErr("Invalid form data") return } @@ -474,14 +479,12 @@ func (ui *AdminUI) handleCrewRemapTier(w http.ResponseWriter, r *http.Request) { to := strings.TrimSpace(r.FormValue("to")) if from == "" || to == "" { - setFlash(w, r, "error", "Both source and target tier are required") - http.Redirect(w, r, "/admin#crew", http.StatusFound) + renderErr("Both source and target tier are required") return } if ui.quotaMgr == nil || !ui.quotaMgr.IsEnabled() { - setFlash(w, r, "error", "Quotas are not enabled on this hold") - http.Redirect(w, r, "/admin#crew", http.StatusFound) + renderErr("Quotas are not enabled on this hold") return } @@ -495,60 +498,86 @@ func (ui *AdminUI) handleCrewRemapTier(w http.ResponseWriter, r *http.Request) { } } if !validTo { - setFlash(w, r, "error", fmt.Sprintf("Unknown target tier %q", to)) - http.Redirect(w, r, "/admin#crew", http.StatusFound) + renderErr(fmt.Sprintf("Unknown target tier %q", to)) return } + // List crew synchronously (fast) so we can report load failures inline and + // size the progress bar before detaching. members, err := ui.pds.ListCrewMembers(ctx) if err != nil { slog.Error("Failed to list crew members for tier remap", "error", err) - setFlash(w, r, "error", "Failed to load crew: "+err.Error()) - http.Redirect(w, r, "/admin#crew", http.StatusFound) + renderErr("Failed to load crew: " + err.Error()) return } - var ok, failed int + total := 0 for _, member := range members { - if member.Record.Tier != from { - continue + if member.Record.Tier == from { + total++ } - if err := ui.pds.UpdateCrewMemberTier(ctx, member.Record.Member, to); err != nil { - slog.Warn("Failed to remap crew tier", - "did", member.Record.Member, - "from", from, - "to", to, - "error", err) - failed++ - continue - } - ok++ - // Throttle firehose events. Matches the convention in - // pkg/hold/gc/gc.go (image config backfill). - time.Sleep(200 * time.Millisecond) + } + if total == 0 { + ui.renderTemplate(w, "partials/job_result.html", jobResult{ + Message: fmt.Sprintf("No crew were on tier %q.", from), + }) + return } session := getSessionFromContext(ctx) - slog.Info("Bulk crew tier remap", - "from", from, - "to", to, - "updated", ok, - "failed", failed, - "by", func() string { - if session != nil { - return session.DID - } - return "" - }()) - - if ok == 0 && failed == 0 { - setFlash(w, r, "info", fmt.Sprintf("No crew were on tier %q", from)) - } else if failed > 0 { - setFlash(w, r, "warning", fmt.Sprintf("Remapped %d crew from %q → %q (%d failed, see logs)", ok, from, to, failed)) - } else { - setFlash(w, r, "success", fmt.Sprintf("Remapped %d crew from %q → %q", ok, from, to)) + byDID := "" + if session != nil { + byDID = session.DID } - http.Redirect(w, r, "/admin#crew", http.StatusFound) + + // The per-member UpdateCrewMemberTier loop runs detached so it survives the + // reverse-proxy timeout that was cancelling it mid-run (~50 of N before a + // 504, leaving the rest as "context canceled"). + ui.startJob("crew-remap-tier", "Remapping crew tier", + "partials/job_result.html", 10*time.Minute, + func(jobCtx context.Context, progress func(jobProgress)) (any, error) { + var done, ok, failed int + for _, member := range members { + if member.Record.Tier != from { + continue + } + done++ + progress(jobProgress{ + Done: done, + Total: total, + Message: fmt.Sprintf("Remapping %s", member.Record.Member), + }) + if err := ui.pds.UpdateCrewMemberTier(jobCtx, member.Record.Member, to); err != nil { + slog.Warn("Failed to remap crew tier", + "did", member.Record.Member, + "from", from, + "to", to, + "error", err) + failed++ + continue + } + ok++ + // Throttle firehose events. Matches the convention in + // pkg/hold/gc/gc.go (image config backfill). + time.Sleep(200 * time.Millisecond) + } + + slog.Info("Bulk crew tier remap", + "from", from, "to", to, "updated", ok, "failed", failed, "by", byDID) + + msg := fmt.Sprintf("Remapped %d crew from %q → %q", ok, from, to) + if failed > 0 { + msg += fmt.Sprintf(" (%d failed, see logs)", failed) + } + return jobResult{ + Message: msg, + Failed: failed, + ReloadURL: "/admin/api/tab/crew", + ReloadTarget: "#tab-crew", + }, nil + }) + + ui.renderTemplate(w, "partials/job_progress.html", ui.jobSnapshot("crew-remap-tier")) } // handleCrewDelete removes a crew member diff --git a/pkg/hold/admin/handlers_crew_io.go b/pkg/hold/admin/handlers_crew_io.go index 6a6191f..e283d21 100644 --- a/pkg/hold/admin/handlers_crew_io.go +++ b/pkg/hold/admin/handlers_crew_io.go @@ -1,6 +1,7 @@ package admin import ( + "context" "encoding/json" "fmt" "log/slog" @@ -34,6 +35,15 @@ type importResult struct { Reason string } +// crewImportResult is the job result rendered by partials/crew_import_results.html. +type crewImportResult struct { + Results []importResult + Added int + Skipped int + Errors int + Total int +} + const maxImportSize = 1 << 20 // 1 MB // handleCrewExport exports all crew members as a JSON file download @@ -92,22 +102,28 @@ func (ui *AdminUI) handleCrewImportForm(w http.ResponseWriter, r *http.Request) ui.renderTemplate(w, "pages/crew_import.html", data) } -// handleCrewImport processes an uploaded crew JSON file +// handleCrewImport processes an uploaded crew JSON file. The upload is parsed +// and decoded synchronously (the request body can't be read after the handler +// returns), then the per-entry loop — which does a network handle resolution +// plus a PDS write per member — runs as a detached background job so a large +// file can't 504. The form is htmx-driven, so failures render 200 fragments. func (ui *AdminUI) handleCrewImport(w http.ResponseWriter, r *http.Request) { ctx := r.Context() + renderErr := func(msg string) { + ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{msg}) + } + r.Body = http.MaxBytesReader(w, r.Body, maxImportSize) if err := r.ParseMultipartForm(maxImportSize); err != nil { - setFlash(w, r, "error", "File too large (max 1 MB)") - http.Redirect(w, r, "/admin/crew/import", http.StatusFound) + renderErr("File too large (max 1 MB)") return } file, _, err := r.FormFile("crew_file") if err != nil { - setFlash(w, r, "error", "No file uploaded") - http.Redirect(w, r, "/admin/crew/import", http.StatusFound) + renderErr("No file uploaded") return } defer file.Close() @@ -115,99 +131,103 @@ func (ui *AdminUI) handleCrewImport(w http.ResponseWriter, r *http.Request) { var export crewExportFile dec := json.NewDecoder(file) if err := dec.Decode(&export); err != nil { - setFlash(w, r, "error", "Invalid JSON: "+err.Error()) - http.Redirect(w, r, "/admin/crew/import", http.StatusFound) + renderErr("Invalid JSON: " + err.Error()) return } if export.Version != 1 { - setFlash(w, r, "error", fmt.Sprintf("Unsupported export version: %d (expected 1)", export.Version)) - http.Redirect(w, r, "/admin/crew/import", http.StatusFound) + renderErr(fmt.Sprintf("Unsupported export version: %d (expected 1)", export.Version)) return } if len(export.Crew) == 0 { - setFlash(w, r, "error", "No crew members in file") - http.Redirect(w, r, "/admin/crew/import", http.StatusFound) + renderErr("No crew members in file") return } - var results []importResult - for _, entry := range export.Crew { - result := importResult{DID: entry.DID} - - if !strings.HasPrefix(entry.DID, "did:") { - result.Status = "error" - result.Reason = "Invalid DID format" - results = append(results, result) - continue - } - - // Check if member already exists (O(1) lookup) - _, _, err := ui.pds.GetCrewMemberByDID(ctx, entry.DID) - if err == nil { - result.Status = "skipped" - result.Reason = "Already exists" - results = append(results, result) - continue - } - - role := entry.Role - if role == "" { - role = "member" - } - - // Resolve tier: use entry tier if specified, otherwise default from quota config - tier := entry.Tier - if tier == "" && ui.quotaMgr != nil && ui.quotaMgr.IsEnabled() { - tier = ui.quotaMgr.GetDefaultTier() - } - - if _, err := ui.pds.AddCrewMember(ctx, entry.DID, role, entry.Permissions, tier); err != nil { - result.Status = "error" - result.Reason = err.Error() - results = append(results, result) - continue - } - - result.Status = "added" - result.Handle = resolveHandle(ctx, entry.DID) - results = append(results, result) - } - - var added, skipped, errored int - for _, res := range results { - switch res.Status { - case "added": - added++ - case "skipped": - skipped++ - case "error": - errored++ - } - } - + // export.Crew is fully in memory (request body capped at maxImportSize), so + // it's safe to hand to the detached job. + entries := export.Crew session := getSessionFromContext(ctx) - slog.Info("Crew imported via admin panel", - "added", added, - "skipped", skipped, - "errors", errored, - "by", session.DID) - - data := struct { - PageData - Results []importResult - Added int - Skipped int - Errors int - Total int - }{ - PageData: ui.newPageData(r, "Import Results", "crew"), - Results: results, - Added: added, - Skipped: skipped, - Errors: errored, - Total: len(results), + byDID := "" + if session != nil { + byDID = session.DID } - ui.renderTemplate(w, "pages/crew_import_results.html", data) + + ui.startJob("crew-import", "Importing crew", + "partials/crew_import_results.html", 10*time.Minute, + func(jobCtx context.Context, progress func(jobProgress)) (any, error) { + results := make([]importResult, 0, len(entries)) + for i, entry := range entries { + progress(jobProgress{ + Done: i + 1, + Total: len(entries), + Message: fmt.Sprintf("Importing %s", entry.DID), + }) + + result := importResult{DID: entry.DID} + + if !strings.HasPrefix(entry.DID, "did:") { + result.Status = "error" + result.Reason = "Invalid DID format" + results = append(results, result) + continue + } + + // Check if member already exists (O(1) lookup) + if _, _, err := ui.pds.GetCrewMemberByDID(jobCtx, entry.DID); err == nil { + result.Status = "skipped" + result.Reason = "Already exists" + results = append(results, result) + continue + } + + role := entry.Role + if role == "" { + role = "member" + } + + // Resolve tier: use entry tier if specified, otherwise default from quota config + tier := entry.Tier + if tier == "" && ui.quotaMgr != nil && ui.quotaMgr.IsEnabled() { + tier = ui.quotaMgr.GetDefaultTier() + } + + if _, err := ui.pds.AddCrewMember(jobCtx, entry.DID, role, entry.Permissions, tier); err != nil { + result.Status = "error" + result.Reason = err.Error() + results = append(results, result) + continue + } + + result.Status = "added" + result.Handle = resolveHandle(jobCtx, entry.DID) + results = append(results, result) + } + + var added, skipped, errored int + for _, res := range results { + switch res.Status { + case "added": + added++ + case "skipped": + skipped++ + case "error": + errored++ + } + } + + slog.Info("Crew imported via admin panel", + "added", added, "skipped", skipped, "errors", errored, "by", byDID) + + return crewImportResult{ + Results: results, + Added: added, + Skipped: skipped, + Errors: errored, + Total: len(results), + }, nil + }) + + ui.renderTemplate(w, "partials/job_progress.html", ui.jobSnapshot("crew-import")) } diff --git a/pkg/hold/admin/handlers_scan.go b/pkg/hold/admin/handlers_scan.go index c06dcb5..676be58 100644 --- a/pkg/hold/admin/handlers_scan.go +++ b/pkg/hold/admin/handlers_scan.go @@ -12,14 +12,14 @@ import ( "atcr.io/pkg/hold/pds" ) -// handleScanBackfill kicks off a scan-status backfill in a background -// goroutine and returns a progress fragment that polls -// /admin/api/scan-backfill/status for updates. Idempotent — clicking again -// while a run is in flight just shows the current progress. +// handleScanBackfill kicks off a scan-status backfill as a background job and +// returns a progress fragment that polls /admin/api/jobs/scan-backfill/status. +// Idempotent — clicking again while a run is in flight just shows the current +// progress (startJob returns false). // // Why background: reverse proxies typically cap upstream HTTP timeouts at -// 10–60s, which would cancel a synchronous request mid-loop. Detaching the -// work from the request context lets it run to completion. +// 10–60s, which would cancel a synchronous request mid-loop. The job runs under +// its own detached context (see jobs.go), so it survives the request ending. // // JSON callers (Accept: application/json) get a synchronous run instead — // useful for curl + scripting. @@ -50,119 +50,36 @@ func (ui *AdminUI) handleScanBackfill(w http.ResponseWriter, r *http.Request) { return } - // HTML path — kick off the background run if one isn't already going. - st := &ui.scanBackfill - st.mu.Lock() - alreadyRunning := st.running - if !alreadyRunning { - st.running = true - st.startedAt = time.Now() - st.current = &pds.ScanBackfillResult{} - st.result = nil - st.err = "" - } - st.mu.Unlock() + started := ui.startJob("scan-backfill", "Backfilling scan records", + "partials/scan_backfill_result.html", 10*time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + res, err := ui.pds.BackfillScanStatus(ctx, scanBackfillLogger, func(snap *pds.ScanBackfillResult) { + progress(jobProgress{ + Done: snap.Scanned, + Message: fmt.Sprintf("Scanned %d · rewrites %d (%d skipped, %d failed)", + snap.Scanned, snap.Rewritten, snap.MarkedSkipped, snap.MarkedFailed), + }) + }) + if err != nil { + return nil, err + } + slog.Info("scan-status backfill complete", + "scanned", res.Scanned, + "already_tagged", res.AlreadyTagged, + "marked_skipped", res.MarkedSkipped, + "marked_failed", res.MarkedFailed, + "rewritten", res.Rewritten, + ) + return res, nil + }) - if !alreadyRunning { + if started { slog.Info("scan-status backfill started via admin panel", "by", session.DID) - go ui.runScanBackfill() } else { slog.Debug("scan-status backfill already in progress; returning current state") } - ui.renderTemplate(w, "partials/scan_backfill_progress.html", ui.snapshotScanBackfill()) -} - -// handleScanBackfillStatus is polled by the progress fragment. Returns the -// progress fragment again if running, the result fragment when done, or an -// error fragment if something went wrong. -func (ui *AdminUI) handleScanBackfillStatus(w http.ResponseWriter, r *http.Request) { - snap := ui.snapshotScanBackfill() - if snap.Running { - ui.renderTemplate(w, "partials/scan_backfill_progress.html", snap) - return - } - if snap.Error != "" { - ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{snap.Error}) - return - } - if snap.Result == nil { - // Initial state, before any run — render an empty placeholder. - _, _ = w.Write([]byte("")) - return - } - ui.renderTemplate(w, "partials/scan_backfill_result.html", snap.Result) -} - -// runScanBackfill is the goroutine body. Updates the shared state as the -// backfill progresses and stores the final result (or error) when it ends. -func (ui *AdminUI) runScanBackfill() { - st := &ui.scanBackfill - defer func() { - st.mu.Lock() - st.running = false - st.mu.Unlock() - }() - - // Generous independent timeout — the loop is single-threaded and large - // holds with thousands of legacy records can take a few minutes. - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) - defer cancel() - - res, err := ui.pds.BackfillScanStatus(ctx, scanBackfillLogger, func(snap *pds.ScanBackfillResult) { - // Copy so we don't keep a pointer the loop will mutate. - c := *snap - st.mu.Lock() - st.current = &c - st.mu.Unlock() - }) - st.mu.Lock() - if err != nil { - st.err = err.Error() - slog.Error("scan-status backfill failed", "error", err) - } else { - st.result = res - slog.Info("scan-status backfill complete", - "scanned", res.Scanned, - "already_tagged", res.AlreadyTagged, - "marked_skipped", res.MarkedSkipped, - "marked_failed", res.MarkedFailed, - "rewritten", res.Rewritten, - ) - } - st.mu.Unlock() -} - -// scanBackfillSnapshot is the shape exposed to templates. -type scanBackfillSnapshot struct { - Running bool - StartedAt time.Time - Current *pds.ScanBackfillResult // populated while running - Result *pds.ScanBackfillResult // populated when complete - Error string -} - -// snapshotScanBackfill returns a copy of the current state — safe to render -// without holding the mutex. -func (ui *AdminUI) snapshotScanBackfill() scanBackfillSnapshot { - st := &ui.scanBackfill - st.mu.Lock() - defer st.mu.Unlock() - - snap := scanBackfillSnapshot{ - Running: st.running, - StartedAt: st.startedAt, - Error: st.err, - } - if st.current != nil { - c := *st.current - snap.Current = &c - } - if st.result != nil { - r := *st.result - snap.Result = &r - } - return snap + ui.renderTemplate(w, "partials/job_progress.html", ui.jobSnapshot("scan-backfill")) } // scanBackfillLogger formats the printf-style messages from BackfillScanStatus diff --git a/pkg/hold/admin/jobs.go b/pkg/hold/admin/jobs.go new file mode 100644 index 0000000..a6be9bc --- /dev/null +++ b/pkg/hold/admin/jobs.go @@ -0,0 +1,183 @@ +package admin + +import ( + "context" + "net/http" + "sync" + "time" + + "github.com/go-chi/chi/v5" +) + +// Long-running admin operations (bulk crew tier remaps, crew imports, +// scan-record backfills) must NOT run as a synchronous loop bound to the HTTP +// request context. A reverse proxy times out such a request around the 10s mark +// and cancels r.Context(), which aborts the work mid-flight ("context canceled" +// from the blockstore). Instead they run here as a detached background job: the +// kickoff handler returns a progress fragment immediately, the work runs under +// its own context.Background() timeout, and the page polls +// /admin/api/jobs/{key}/status until it finishes. +// +// This is the same shape hand-rolled in gc.startBackground; the admin package +// keeps its own copy because gc must not import admin. + +// jobProgress is the live progress of a running job. It is a value type copied +// under the job's lock — never hand a pointer the job loop mutates to a template. +type jobProgress struct { + Done int + Total int // 0 = indeterminate (spinner only, no progress bar) + Message string +} + +// jobResult is the generic success payload rendered by partials/job_result.html. +// Jobs with richer output (crew import, scan backfill) register their own result +// template and return a different struct instead. +type jobResult struct { + Message string // summary line + Failed int // >0 renders the alert as a warning rather than success + ReloadURL string // optional htmx affordance to reload a tab after the job + ReloadTarget string // CSS selector the ReloadURL swaps into +} + +// jobState is the state machine for one background job, keyed by a stable string +// (e.g. "crew-remap-tier"). Only one run per key at a time. +type jobState struct { + mu sync.Mutex + title string + resultTemplate string + started bool + running bool + startedAt time.Time + progress jobProgress + result any + err string +} + +// jobRegistry holds one jobState per key. The zero value is ready to use. +type jobRegistry struct { + mu sync.Mutex + jobs map[string]*jobState +} + +// get returns the jobState for key, creating an empty one on first access. +func (r *jobRegistry) get(key string) *jobState { + r.mu.Lock() + defer r.mu.Unlock() + if r.jobs == nil { + r.jobs = make(map[string]*jobState) + } + st, ok := r.jobs[key] + if !ok { + st = &jobState{} + r.jobs[key] = st + } + return st +} + +// jobSnapshot is the read-only view rendered to templates. +type jobSnapshot struct { + Key string + Title string + ResultTemplate string + Started bool + Running bool + StartedAt time.Time + Progress jobProgress + Result any + Error string +} + +// startJob launches fn in a detached goroutine under its own timeout and returns +// true. If a job with this key is already running it returns false and leaves the +// in-flight run untouched (the caller should just render the current snapshot). +// +// fn publishes progress via the passed callback and returns a result value +// (rendered by resultTemplate) or an error (rendered by partials/gc_error.html). +// fn must use the ctx it is given — that ctx carries the detached timeout, not +// the request deadline. +func (ui *AdminUI) startJob(key, title, resultTemplate string, timeout time.Duration, + fn func(ctx context.Context, progress func(jobProgress)) (any, error)) bool { + st := ui.jobs.get(key) + + st.mu.Lock() + if st.running { + st.mu.Unlock() + return false + } + st.running = true + st.started = true + st.startedAt = time.Now() + st.title = title + st.resultTemplate = resultTemplate + st.progress = jobProgress{} + st.result = nil + st.err = "" + st.mu.Unlock() + + publish := func(p jobProgress) { + st.mu.Lock() + st.progress = p + st.mu.Unlock() + } + + go func() { + defer func() { + st.mu.Lock() + st.running = false + st.mu.Unlock() + }() + + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + res, err := fn(ctx, publish) + + st.mu.Lock() + if err != nil { + st.err = err.Error() + } else { + st.result = res + } + st.mu.Unlock() + }() + + return true +} + +// jobSnapshot returns a copy of the current state for key — safe to render +// without holding the lock. +func (ui *AdminUI) jobSnapshot(key string) jobSnapshot { + st := ui.jobs.get(key) + st.mu.Lock() + defer st.mu.Unlock() + return jobSnapshot{ + Key: key, + Title: st.title, + ResultTemplate: st.resultTemplate, + Started: st.started, + Running: st.running, + StartedAt: st.startedAt, + Progress: st.progress, + Result: st.result, + Error: st.err, + } +} + +// handleJobStatus is polled by the progress fragment. It renders the progress +// fragment while running, the error fragment on failure, or the job's registered +// result template on success. A never-started key renders an empty body. +func (ui *AdminUI) handleJobStatus(w http.ResponseWriter, r *http.Request) { + key := chi.URLParam(r, "key") + snap := ui.jobSnapshot(key) + + switch { + case !snap.Started: + _, _ = w.Write([]byte("")) + case snap.Running: + ui.renderTemplate(w, "partials/job_progress.html", snap) + case snap.Error != "": + ui.renderTemplate(w, "partials/gc_error.html", struct{ Error string }{snap.Error}) + default: + ui.renderTemplate(w, snap.ResultTemplate, snap.Result) + } +} diff --git a/pkg/hold/admin/jobs_test.go b/pkg/hold/admin/jobs_test.go new file mode 100644 index 0000000..202ae52 --- /dev/null +++ b/pkg/hold/admin/jobs_test.go @@ -0,0 +1,309 @@ +package admin + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/go-chi/chi/v5" +) + +// waitDone polls until the job is finished (Started && !Running). Because the +// goroutine sets running=false in a defer that runs AFTER the result is stored, +// observing !Running guarantees the result/error is visible — no race. +func waitDone(t *testing.T, ui *AdminUI, key string) jobSnapshot { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + snap := ui.jobSnapshot(key) + if snap.Started && !snap.Running { + return snap + } + time.Sleep(time.Millisecond) + } + t.Fatalf("job %q did not finish within deadline", key) + return jobSnapshot{} +} + +func TestStartJob_HappyPath(t *testing.T) { + ui := &AdminUI{} + release := make(chan struct{}) + + started := ui.startJob("k", "Doing thing", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + <-release + return jobResult{Message: "all done"}, nil + }) + if !started { + t.Fatal("startJob returned false for a fresh key") + } + + // While the fn is blocked, the job is observably running. + snap := ui.jobSnapshot("k") + if !snap.Started || !snap.Running { + t.Fatalf("expected started+running, got started=%v running=%v", snap.Started, snap.Running) + } + if snap.Title != "Doing thing" { + t.Errorf("title = %q, want %q", snap.Title, "Doing thing") + } + + close(release) + snap = waitDone(t, ui, "k") + + if snap.Running { + t.Error("job still running after completion") + } + if snap.Error != "" { + t.Errorf("unexpected error: %q", snap.Error) + } + res, ok := snap.Result.(jobResult) + if !ok { + t.Fatalf("result type = %T, want jobResult", snap.Result) + } + if res.Message != "all done" { + t.Errorf("result message = %q, want %q", res.Message, "all done") + } +} + +func TestStartJob_DoubleStartGuard(t *testing.T) { + ui := &AdminUI{} + release := make(chan struct{}) + + if !ui.startJob("k", "first", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + <-release + return jobResult{Message: "first"}, nil + }) { + t.Fatal("first startJob returned false") + } + + // Second start while the first is in flight must be rejected and must not + // clobber the running job's title. + if ui.startJob("k", "second", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + return jobResult{Message: "second"}, nil + }) { + t.Fatal("second startJob returned true while a run was in flight") + } + if snap := ui.jobSnapshot("k"); snap.Title != "first" { + t.Errorf("running job title clobbered: %q", snap.Title) + } + + close(release) + snap := waitDone(t, ui, "k") + if res := snap.Result.(jobResult); res.Message != "first" { + t.Errorf("result message = %q, want from first run", res.Message) + } +} + +func TestStartJob_RerunAfterCompletion(t *testing.T) { + ui := &AdminUI{} + + ui.startJob("k", "run1", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + progress(jobProgress{Done: 5, Total: 5}) + return jobResult{Message: "run1"}, nil + }) + waitDone(t, ui, "k") + + // A second start with the same key succeeds and resets progress/result/err. + release := make(chan struct{}) + if !ui.startJob("k", "run2", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + <-release + return jobResult{Message: "run2"}, nil + }) { + t.Fatal("re-run startJob returned false after completion") + } + snap := ui.jobSnapshot("k") + if snap.Progress != (jobProgress{}) { + t.Errorf("progress not reset on re-run: %+v", snap.Progress) + } + if snap.Result != nil { + t.Errorf("result not reset on re-run: %v", snap.Result) + } + close(release) + waitDone(t, ui, "k") +} + +func TestStartJob_ProgressPublishingIsCopied(t *testing.T) { + ui := &AdminUI{} + publishedOne := make(chan struct{}) + proceed := make(chan struct{}) + publishedTwo := make(chan struct{}) + release := make(chan struct{}) + + ui.startJob("k", "t", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + progress(jobProgress{Done: 1, Total: 3, Message: "one"}) + publishedOne <- struct{}{} + <-proceed // wait until the test has snapshotted the first value + progress(jobProgress{Done: 2, Total: 3, Message: "two"}) + publishedTwo <- struct{}{} + <-release + return jobResult{}, nil + }) + + // Synchronize on the first publish, then snapshot it (fn is parked on proceed). + <-publishedOne + first := ui.jobSnapshot("k") + if first.Progress.Done != 1 || first.Progress.Message != "one" { + t.Fatalf("first progress = %+v", first.Progress) + } + + // Let the job publish again; the earlier snapshot must be an independent copy. + proceed <- struct{}{} + <-publishedTwo + if first.Progress.Done != 1 || first.Progress.Message != "one" { + t.Errorf("earlier snapshot mutated: %+v", first.Progress) + } + second := ui.jobSnapshot("k") + if second.Progress.Done != 2 || second.Progress.Message != "two" { + t.Errorf("second progress = %+v", second.Progress) + } + + close(release) + waitDone(t, ui, "k") +} + +func TestStartJob_ErrorPath(t *testing.T) { + ui := &AdminUI{} + ui.startJob("k", "t", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + return nil, errors.New("boom") + }) + snap := waitDone(t, ui, "k") + if snap.Error != "boom" { + t.Errorf("error = %q, want %q", snap.Error, "boom") + } + if snap.Result != nil { + t.Errorf("result = %v, want nil on error", snap.Result) + } +} + +func TestStartJob_RespectsTimeout(t *testing.T) { + ui := &AdminUI{} + ui.startJob("k", "t", "partials/job_result.html", 10*time.Millisecond, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + <-ctx.Done() // detached timeout fires here, not the request deadline + return nil, ctx.Err() + }) + snap := waitDone(t, ui, "k") + if snap.Error != context.DeadlineExceeded.Error() { + t.Errorf("error = %q, want %q", snap.Error, context.DeadlineExceeded.Error()) + } +} + +func TestJobSnapshot_UnknownKey(t *testing.T) { + ui := &AdminUI{} + snap := ui.jobSnapshot("never-run") + if snap.Started || snap.Running { + t.Errorf("unknown key reported started=%v running=%v", snap.Started, snap.Running) + } +} + +func TestStartJob_DistinctKeysAreIndependent(t *testing.T) { + ui := &AdminUI{} + releaseA := make(chan struct{}) + + ui.startJob("a", "A", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + <-releaseA + return jobResult{Message: "a"}, nil + }) + ui.startJob("b", "B", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + return jobResult{Message: "b"}, nil + }) + + // b finishes while a is still blocked. + bSnap := waitDone(t, ui, "b") + if bSnap.Result.(jobResult).Message != "b" { + t.Errorf("b result = %v", bSnap.Result) + } + if aSnap := ui.jobSnapshot("a"); !aSnap.Running { + t.Error("job a should still be running while b finished") + } + + close(releaseA) + aSnap := waitDone(t, ui, "a") + if aSnap.Result.(jobResult).Message != "a" { + t.Errorf("a result = %v", aSnap.Result) + } +} + +// TestStartJob_OutlivesCaller is the core regression: the work must complete +// even after the calling scope (the HTTP handler) has returned. +func TestStartJob_OutlivesCaller(t *testing.T) { + ui := &AdminUI{} + + // kickoff mimics a handler that returns immediately after starting the job. + kickoff := func() { + ui.startJob("k", "t", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + time.Sleep(20 * time.Millisecond) + return jobResult{Message: "finished after handler returned"}, nil + }) + } + kickoff() // handler scope ends here + + snap := waitDone(t, ui, "k") + if snap.Result.(jobResult).Message != "finished after handler returned" { + t.Errorf("job did not complete after caller returned: %v", snap.Result) + } +} + +func TestHandleJobStatus_Dispatch(t *testing.T) { + tmpls, err := parseTemplates() + if err != nil { + t.Fatalf("parseTemplates: %v", err) + } + ui := &AdminUI{templates: tmpls} + + status := func(key string) (int, string) { + req := httptest.NewRequest(http.MethodGet, "/admin/api/jobs/"+key+"/status", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("key", key) + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + rec := httptest.NewRecorder() + ui.handleJobStatus(rec, req) + return rec.Code, rec.Body.String() + } + + // not-started -> empty body + if code, body := status("nope"); code != http.StatusOK || strings.TrimSpace(body) != "" { + t.Errorf("not-started: code=%d body=%q", code, body) + } + + // running -> progress fragment (polls the status endpoint) + release := make(chan struct{}) + ui.startJob("running", "Working", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + <-release + return jobResult{Message: "done"}, nil + }) + if _, body := status("running"); !strings.Contains(body, "/admin/api/jobs/running/status") { + t.Errorf("running fragment missing poll URL: %q", body) + } + close(release) + waitDone(t, ui, "running") + + // done -> result template + if _, body := status("running"); !strings.Contains(body, "done") { + t.Errorf("result fragment missing message: %q", body) + } + + // error -> gc_error fragment + ui.startJob("failed", "Working", "partials/job_result.html", time.Minute, + func(ctx context.Context, progress func(jobProgress)) (any, error) { + return nil, errors.New("kaput") + }) + waitDone(t, ui, "failed") + if _, body := status("failed"); !strings.Contains(body, "alert-error") || !strings.Contains(body, "kaput") { + t.Errorf("error fragment wrong: %q", body) + } +} diff --git a/pkg/hold/admin/templates/pages/crew_import.html b/pkg/hold/admin/templates/pages/crew_import.html index 547e674..9a15d65 100644 --- a/pkg/hold/admin/templates/pages/crew_import.html +++ b/pkg/hold/admin/templates/pages/crew_import.html @@ -11,7 +11,11 @@ {{define "page-content"}}
-
+ {{ csrfInput .CSRFToken }}
@@ -37,6 +41,8 @@ Cancel
+ +
{{end}} diff --git a/pkg/hold/admin/templates/pages/crew_import_results.html b/pkg/hold/admin/templates/partials/crew_import_results.html similarity index 85% rename from pkg/hold/admin/templates/pages/crew_import_results.html rename to pkg/hold/admin/templates/partials/crew_import_results.html index e8f4d11..eb2cadc 100644 --- a/pkg/hold/admin/templates/pages/crew_import_results.html +++ b/pkg/hold/admin/templates/partials/crew_import_results.html @@ -1,14 +1,4 @@ -{{define "page-header"}} - -{{end}} - -{{define "page-content"}} +{{define "partials/crew_import_results.html"}}
Added
@@ -63,6 +53,6 @@ {{end}} diff --git a/pkg/hold/admin/templates/partials/job_progress.html b/pkg/hold/admin/templates/partials/job_progress.html new file mode 100644 index 0000000..0a4d185 --- /dev/null +++ b/pkg/hold/admin/templates/partials/job_progress.html @@ -0,0 +1,22 @@ +{{define "partials/job_progress.html"}} +
+ +
+

{{ .Title }}...

+ {{ if .Progress.Message }} +

+ {{ .Progress.Message }} + {{ if .Progress.Total }}({{ .Progress.Done }}/{{ .Progress.Total }}){{ else if .Progress.Done }}({{ .Progress.Done }}){{ end }} +

+ {{ else }} +

Starting...

+ {{ end }} + {{ if .Progress.Total }} + + {{ end }} +
+
+{{end}} diff --git a/pkg/hold/admin/templates/partials/job_result.html b/pkg/hold/admin/templates/partials/job_result.html new file mode 100644 index 0000000..652ac18 --- /dev/null +++ b/pkg/hold/admin/templates/partials/job_result.html @@ -0,0 +1,15 @@ +{{define "partials/job_result.html"}} +
+ {{ if .Failed }}{{ icon "triangle-alert" "size-5" }}{{ else }}{{ icon "check-circle" "size-5 shrink-0" }}{{ end }} + {{ .Message }} + {{ if .ReloadURL }} + + {{ end }} +
+{{end}} diff --git a/pkg/hold/admin/templates/partials/scan_backfill_progress.html b/pkg/hold/admin/templates/partials/scan_backfill_progress.html deleted file mode 100644 index edcf72d..0000000 --- a/pkg/hold/admin/templates/partials/scan_backfill_progress.html +++ /dev/null @@ -1,20 +0,0 @@ -{{define "partials/scan_backfill_progress.html"}} -
- -
-

Backfilling scan records...

- {{ if .Current }} -

- Scanned {{ .Current.Scanned }} records · - rewrites: {{ .Current.Rewritten }} - ({{ .Current.MarkedSkipped }} skipped, {{ .Current.MarkedFailed }} failed) -

- {{ else }} -

Starting...

- {{ end }} -
-
-{{end}} diff --git a/pkg/hold/admin/templates/partials/tab_crew.html b/pkg/hold/admin/templates/partials/tab_crew.html index cb123b6..abed99f 100644 --- a/pkg/hold/admin/templates/partials/tab_crew.html +++ b/pkg/hold/admin/templates/partials/tab_crew.html @@ -48,7 +48,10 @@ {{.Name}} {{.Count}} -
+ {{ csrfInput $.CSRFToken }}
+
{{end}}