From ab4a4ebf9de1362b613463581448aa3b76285fba Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Fri, 5 Jun 2026 20:57:02 -0500 Subject: [PATCH] admin panel long running imrovements, billing fixes, ui cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Multiple registry domains + per-user domain preference The biggest feature. The appview can serve several registry domains (e.g. buoy.cr, atcr.io), and users can now pick which one shows up in their pull/push commands. - Lexicon/record: adds registryDomain (and documents ociClient) to the sailor profile (lexicons/.../profile.json, pkg/atproto/lexicon.go). - DB: new registry_domain column on users (schema.sql + migration 0027), with GetUserByDID/Handle reads, UpdateUserRegistryDomain writer, and Jetstream caching it on profile updates (writes unconditionally so clearing propagates). - UI/handlers: new UpdateRegistryDomainHandler + /api/profile/registry-domain route, a + {{ if gt (len .RegistryDomains) 1 }} +
+
+ +

Sets the registry domain shown in pull and push commands across the site. All domains work, this only changes what is displayed.

+
+ {{ $pref := .Profile.RegistryDomain }} + +
+ {{ 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"}} -
-

Import Results

- - {{ icon "arrow-left" "size-4" }} - Back to Crew - -
-{{end}} - -{{define "page-content"}} +{{define "partials/crew_import_results.html"}}
Added
@@ -63,6 +53,6 @@
Done - Import Another + Import another
{{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}}