admin panel long running imrovements, billing fixes, ui cleanup

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 <select> in the user settings panel (only shown when >1 domain configured), and resolveRegistryURL()
which falls back to the primary domain if the user's pref is stale/removed. Tests added for all of it.

2. default_hold_did removed → first managed_holds entry is the default

Consolidates two overlapping config fields into one. ServerConfig.DefaultHoldDID is gone;
PrimaryHoldDID() now returns managed_holds[0]. managed_holds is now REQUIRED.
Updated in config, validation, server wiring, test harness, example YAML, and the deploy template.

3. Admin long-running operations → generic background-job framework

New pkg/hold/admin/jobs.go introduces a reusable startJob/jobRegistry pattern
 (a detached context.Background() job + a /admin/api/jobs/{key}/status polling endpoint).
This replaces the bespoke scan-backfill goroutine state machine, and now also wraps crew tier remap and crew import
all three previously looped synchronously on the request context and got 504'd/cancelled mid-run by the reverse proxy.
 Forms switched from POST-redirect to htmx fragments (job_progress.html, job_result.html, crew_import_results.html)
 the old crew_import_results.html page and scan_backfill_progress.html partial were deleted.
This is also captured as a new rule in CLAUDE.md.

4. Cascade-delete manifest on last-tag deletion

DeleteTagHandler now, after removing the last tag pointing to a digest, cascade-deletes the manifest itself
 (PDS + DB + hold blob purge) — but only if it's not a child of a manifest list (multi-arch parent).
 New GetTagDigest and ShouldCascadeDeleteManifest queries back it, plus cascade_delete_test.go.
 Also switches tag rkey computation to the atproto.RepositoryTagToRKey helper.

5. Billing simplification

Drops the OwnerBadge config option (hold-owner supporter badge).
The user-profile template no longer special-cases an "owner" badge value (only "Captain").
Example tiers renamed to the nautical scheme (deckhand/bosun/quartermaster).

6. Build/deploy: go generate always runs via Make

make generate is now a phony target that always runs go generate ./... (regenerating cbor_gen, icon sprites, etc.),
 and build-trixie depends on it. The deploy tooling (provision.go/update.go)
drops its own runGenerate calls since the Makefile handles it.

7. New cmd/firehose-tap tool (untracked)

A standalone CLI that subscribes to a com.atproto.sync.subscribeRepos endpoint and pretty-prints events,
with emphasis on Sync 1.1 compliance fields (per-op prev CIDs, commit prevData) and a --validate CI mode.
Fits with the recent "more sync1.1 compliant" commit.
This commit is contained in:
Evan Jarrett
2026-06-05 20:57:25 -05:00
parent efabb677e4
commit ab4a4ebf9d
43 changed files with 1768 additions and 387 deletions
+1
View File
@@ -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
+7 -2
View File
@@ -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 \
+395
View File
@@ -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] <subscribeRepos-endpoint>\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
}
+3 -7
View File
@@ -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
+2 -1
View File
@@ -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
-7
View File
@@ -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)
+2 -16
View File
@@ -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
+2 -1
View File
@@ -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
+1 -1
View File
@@ -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
+11
View File
@@ -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",
+15 -10
View File
@@ -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
+4 -4
View File
@@ -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" {
+199
View File
@@ -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")
}
}
@@ -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 '';
+1
View File
@@ -10,6 +10,7 @@ type User struct {
Avatar string
DefaultHoldDID string
OciClient string
RegistryDomain string
LastSeen time.Time
}
+52 -6
View File
@@ -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, &registryDomain, &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, &registryDomain, &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
+58
View File
@@ -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{})
+1
View File
@@ -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)
);
+4 -3
View File
@@ -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
+14 -2
View File
@@ -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,
+28
View File
@@ -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
+49 -7
View File
@@ -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)
}
+67 -10
View File
@@ -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
+5
View File
@@ -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
+41
View File
@@ -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()
+4 -1
View File
@@ -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
+3 -2
View File
@@ -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)
+1 -1
View File
@@ -33,7 +33,7 @@
{{ end }}
<div class="flex flex-wrap items-center justify-center gap-2 min-w-0">
<h1 class="text-2xl md:text-3xl font-display font-bold tracking-tight break-all min-w-0">{{ .ViewedUser.Handle }}</h1>
{{ if or (eq .SupporterBadge "Captain") (eq .SupporterBadge "owner") }}
{{ if eq .SupporterBadge "Captain" }}
<span class="badge badge-sm supporter-badge-owner">{{ .SupporterBadge }}</span>
{{ else if .SupporterBadge }}
<span class="badge badge-sm supporter-badge">{{ .SupporterBadge }}</span>
@@ -5,13 +5,13 @@
<p class="text-base-content/70 mt-1">Customize your experience across the site.</p>
</div>
<div class="flex items-center gap-4">
<div>
<div class="flex flex-col gap-2 sm:flex-row sm:items-center sm:justify-between sm:gap-6">
<div class="sm:flex-1 sm:min-w-0">
<label for="oci-client-select" class="text-sm font-medium">Preferred client</label>
<p id="oci-client-hint" class="text-xs text-base-content/70">Sets the pull command shown on repository pages. Choose <em>Image reference only</em> to copy without a command prefix.</p>
</div>
{{ $oci := .Profile.OciClient }}
<select id="oci-client-select" aria-describedby="oci-client-hint" class="select select-sm select-bordered min-w-40"
<select id="oci-client-select" aria-describedby="oci-client-hint" class="select select-sm select-bordered w-full sm:w-56 sm:shrink-0"
name="oci_client"
hx-post="/api/profile/oci-client"
hx-trigger="change"
@@ -25,6 +25,25 @@
</select>
</div>
{{ if gt (len .RegistryDomains) 1 }}
<div class="flex flex-col gap-2 sm:flex-row sm:items-center sm:justify-between sm:gap-6">
<div class="sm:flex-1 sm:min-w-0">
<label for="registry-domain-select" class="text-sm font-medium">Registry domain</label>
<p id="registry-domain-hint" class="text-xs text-base-content/70">Sets the registry domain shown in pull and push commands across the site. All domains work, this only changes what is displayed.</p>
</div>
{{ $pref := .Profile.RegistryDomain }}
<select id="registry-domain-select" aria-describedby="registry-domain-hint" class="select select-sm select-bordered w-full sm:w-56 sm:shrink-0"
name="registry_domain"
hx-post="/api/profile/registry-domain"
hx-trigger="change"
hx-swap="none">
{{ range $i, $d := .RegistryDomains }}
<option value="{{ $d }}"{{ if or (eq $d $pref) (and (eq $i 0) (eq $pref "")) }} selected{{ end }}>{{ $d }}</option>
{{ end }}
</select>
</div>
{{ end }}
{{ if .AIAdvisorEnabled }}
<div class="divider my-2"></div>
<div class="flex items-start gap-3">
@@ -44,8 +63,8 @@
<div>
<span class="font-medium text-base-content/50">AI Image Advisor</span>
<p class="text-xs text-base-content/70">Analyze your container images for optimization suggestions using AI.</p>
<p class="text-xs text-primary mt-1">
<a href="/settings/billing">Upgrade your plan</a> to enable this feature.
<p class="text-xs text-base-content/70 mt-1">
<a href="/settings/billing" class="text-primary link link-hover">Upgrade your plan</a> to enable this feature.
</p>
</div>
{{ end }}
+4
View File
@@ -356,6 +356,10 @@ type SailorProfileRecord struct {
// "none" means image reference only (no `<client> 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"`
-3
View File
@@ -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.
+12 -18
View File
@@ -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)
+75 -46
View File
@@ -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
+107 -87
View File
@@ -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"))
}
+30 -113
View File
@@ -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
// 1060s, which would cancel a synchronous request mid-loop. Detaching the
// work from the request context lets it run to completion.
// 1060s, 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
+183
View File
@@ -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)
}
}
+309
View File
@@ -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)
}
}
@@ -11,7 +11,11 @@
{{define "page-content"}}
<div class="card bg-base-100 shadow-sm">
<div class="card-body">
<form action="/admin/crew/import" method="POST" enctype="multipart/form-data" class="max-w-lg">
<form hx-post="/admin/crew/import"
hx-encoding="multipart/form-data"
hx-target="#import-status"
hx-swap="innerHTML"
class="max-w-lg">
{{ csrfInput .CSRFToken }}
<fieldset class="fieldset mb-6">
<label class="fieldset-label font-medium" for="crew_file">Crew Export File</label>
@@ -37,6 +41,8 @@
<a href="/admin#crew" class="btn btn-ghost">Cancel</a>
</div>
</form>
<div id="import-status" class="mt-6"></div>
</div>
</div>
{{end}}
@@ -1,14 +1,4 @@
{{define "page-header"}}
<div class="flex justify-between items-center min-h-12 mb-6">
<h1 class="text-2xl font-bold">Import Results</h1>
<a href="/admin#crew" class="btn btn-ghost gap-2">
{{ icon "arrow-left" "size-4" }}
Back to Crew
</a>
</div>
{{end}}
{{define "page-content"}}
{{define "partials/crew_import_results.html"}}
<div class="grid grid-cols-1 sm:grid-cols-3 gap-4 mb-6">
<div class="stat bg-base-100 shadow-sm rounded-lg">
<div class="stat-title">Added</div>
@@ -63,6 +53,6 @@
<div class="flex gap-3 mt-6">
<a href="/admin#crew" class="btn btn-primary">Done</a>
<a href="/admin/crew/import" class="btn btn-ghost">Import Another</a>
<a href="/admin/crew/import" class="btn btn-ghost">Import another</a>
</div>
{{end}}
@@ -0,0 +1,22 @@
{{define "partials/job_progress.html"}}
<div hx-get="/admin/api/jobs/{{.Key}}/status"
hx-trigger="load delay:1s"
hx-swap="outerHTML"
class="alert alert-info">
<span class="loading loading-spinner loading-sm" aria-hidden="true"></span>
<div class="text-sm w-full">
<p class="font-medium">{{ .Title }}...</p>
{{ if .Progress.Message }}
<p class="text-base-content/70 mt-1">
{{ .Progress.Message }}
{{ if .Progress.Total }}<span class="font-mono">({{ .Progress.Done }}/{{ .Progress.Total }})</span>{{ else if .Progress.Done }}<span class="font-mono">({{ .Progress.Done }})</span>{{ end }}
</p>
{{ else }}
<p class="text-base-content/70 mt-1">Starting...</p>
{{ end }}
{{ if .Progress.Total }}
<progress class="progress progress-info w-full mt-2" value="{{ .Progress.Done }}" max="{{ .Progress.Total }}"></progress>
{{ end }}
</div>
</div>
{{end}}
@@ -0,0 +1,15 @@
{{define "partials/job_result.html"}}
<div class="alert {{ if .Failed }}alert-warning{{ else }}alert-success{{ end }}" role="status">
{{ if .Failed }}{{ icon "triangle-alert" "size-5" }}{{ else }}{{ icon "check-circle" "size-5 shrink-0" }}{{ end }}
<span>{{ .Message }}</span>
{{ if .ReloadURL }}
<button type="button" class="btn btn-sm btn-ghost"
hx-get="{{ .ReloadURL }}"
hx-target="{{ .ReloadTarget }}"
hx-swap="innerHTML">
{{ icon "refresh-cw" "size-4" }}
Reload
</button>
{{ end }}
</div>
{{end}}
@@ -1,20 +0,0 @@
{{define "partials/scan_backfill_progress.html"}}
<div hx-get="/admin/api/scan-backfill/status"
hx-trigger="load delay:1s"
hx-swap="outerHTML"
class="alert alert-info">
<span class="loading loading-spinner loading-sm" aria-hidden="true"></span>
<div class="text-sm">
<p class="font-medium">Backfilling scan records...</p>
{{ if .Current }}
<p class="text-base-content/70 mt-1">
Scanned <span class="font-mono">{{ .Current.Scanned }}</span> records ·
rewrites: <span class="font-mono">{{ .Current.Rewritten }}</span>
({{ .Current.MarkedSkipped }} skipped, {{ .Current.MarkedFailed }} failed)
</p>
{{ else }}
<p class="text-base-content/70 mt-1">Starting...</p>
{{ end }}
</div>
</div>
{{end}}
@@ -48,7 +48,10 @@
<td><code class="text-sm">{{.Name}}</code></td>
<td class="text-right tabular-nums">{{.Count}}</td>
<td colspan="2">
<form action="/admin/crew/remap-tier" method="POST" class="flex gap-2 items-center">
<form hx-post="/admin/crew/remap-tier"
hx-target="#remap-status-{{.Name}}"
hx-swap="innerHTML"
class="flex gap-2 items-center">
{{ csrfInput $.CSRFToken }}
<input type="hidden" name="from" value="{{.Name}}">
<select name="to" required class="select select-bordered select-sm w-full max-w-xs">
@@ -59,6 +62,7 @@
</select>
<button type="submit" class="btn btn-warning btn-sm">Apply</button>
</form>
<div id="remap-status-{{.Name}}" class="mt-2"></div>
</td>
</tr>
{{end}}