update scanner, fix tests, fix dockerfile, move keys to db instead of flat files for appview

Dockerfile.hold

@@ -19,7 +19,7 @@ COPY . .
 
 # Build frontend assets (Tailwind CSS, JS bundle, SVG icons)
 RUN npm ci
-RUN npm run css:build && npm run css:copy-hold && npm run js:build:hold && npm run icons:build
+RUN go generate ./...
 
 # Conditionally add billing tag based on build arg
 RUN if [ "$BILLING_ENABLED" = "true" ]; then \

pkg/appview/config.go

@@ -312,7 +312,6 @@ func buildDistributionConfig(cfg *Config, v *viper.Viper) (*configuration.Config
 			"service":        cfg.Auth.ServiceName,
 			"issuer":         cfg.Auth.ServiceName,
 			"rootcertbundle": cfg.Auth.CertPath,
-			"privatekey":     cfg.Auth.KeyPath,
 			"expiration":     int(cfg.Auth.TokenExpiration.Seconds()),
 		},
 	}

pkg/appview/crypto_keys.go

@@ -0,0 +1,182 @@
+package appview
+
+import (
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"database/sql"
+	"encoding/pem"
+	"fmt"
+	"log/slog"
+	"math/big"
+	"os"
+	"path/filepath"
+	"time"
+
+	"atcr.io/pkg/appview/db"
+	"github.com/bluesky-social/indigo/atproto/atcrypto"
+)
+
+// loadOAuthKey loads the OAuth P-256 key with priority: DB → file → generate.
+// Keys loaded from file or newly generated are stored in the DB.
+func loadOAuthKey(database *sql.DB, keyPath string) (*atcrypto.PrivateKeyP256, error) {
+	// Try database first
+	data, err := db.GetCryptoKey(database, "oauth_p256")
+	if err != nil {
+		return nil, fmt.Errorf("failed to query crypto_keys: %w", err)
+	}
+	if data != nil {
+		key, err := atcrypto.ParsePrivateBytesP256(data)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse OAuth key from database: %w", err)
+		}
+		slog.Info("Loaded OAuth P-256 key from database")
+		return key, nil
+	}
+
+	// Try file fallback
+	if keyPath != "" {
+		if fileData, err := os.ReadFile(keyPath); err == nil {
+			key, err := atcrypto.ParsePrivateBytesP256(fileData)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse OAuth key from file %s: %w", keyPath, err)
+			}
+			// Migrate to database
+			if err := db.PutCryptoKey(database, "oauth_p256", fileData); err != nil {
+				return nil, fmt.Errorf("failed to store OAuth key in database: %w", err)
+			}
+			slog.Info("Migrated OAuth P-256 key from file to database", "path", keyPath)
+			return key, nil
+		}
+	}
+
+	// Generate new key
+	p256Key, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		return nil, fmt.Errorf("failed to generate OAuth P-256 key: %w", err)
+	}
+
+	keyBytes := p256Key.Bytes()
+	if err := db.PutCryptoKey(database, "oauth_p256", keyBytes); err != nil {
+		return nil, fmt.Errorf("failed to store generated OAuth key in database: %w", err)
+	}
+	slog.Info("Generated new OAuth P-256 key and stored in database")
+
+	return p256Key, nil
+}
+
+// loadJWTKeyAndCert loads the JWT RSA key from DB (with file fallback) and generates
+// a self-signed certificate. The cert is always regenerated and written to certPath
+// on disk because the distribution library reads it via os.Open().
+func loadJWTKeyAndCert(database *sql.DB, keyPath, certPath string) (*rsa.PrivateKey, []byte, error) {
+	rsaKey, err := loadRSAKey(database, keyPath)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Generate cert and write to disk for distribution library
+	certDER, err := generateAndWriteCert(rsaKey, certPath)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return rsaKey, certDER, nil
+}
+
+// loadRSAKey loads the RSA private key with priority: DB → file → generate.
+func loadRSAKey(database *sql.DB, keyPath string) (*rsa.PrivateKey, error) {
+	// Try database first
+	data, err := db.GetCryptoKey(database, "jwt_rsa")
+	if err != nil {
+		return nil, fmt.Errorf("failed to query crypto_keys: %w", err)
+	}
+	if data != nil {
+		key, err := parseRSAKeyPEM(data)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse RSA key from database: %w", err)
+		}
+		slog.Info("Loaded JWT RSA key from database")
+		return key, nil
+	}
+
+	// Try file fallback
+	if keyPath != "" {
+		if fileData, err := os.ReadFile(keyPath); err == nil {
+			key, err := parseRSAKeyPEM(fileData)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse RSA key from file %s: %w", keyPath, err)
+			}
+			// Migrate to database
+			if err := db.PutCryptoKey(database, "jwt_rsa", fileData); err != nil {
+				return nil, fmt.Errorf("failed to store RSA key in database: %w", err)
+			}
+			slog.Info("Migrated JWT RSA key from file to database", "path", keyPath)
+			return key, nil
+		}
+	}
+
+	// Generate new key
+	rsaKey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		return nil, fmt.Errorf("failed to generate RSA key: %w", err)
+	}
+
+	keyPEM := pem.EncodeToMemory(&pem.Block{
+		Type:  "RSA PRIVATE KEY",
+		Bytes: x509.MarshalPKCS1PrivateKey(rsaKey),
+	})
+	if err := db.PutCryptoKey(database, "jwt_rsa", keyPEM); err != nil {
+		return nil, fmt.Errorf("failed to store generated RSA key in database: %w", err)
+	}
+	slog.Info("Generated new JWT RSA key and stored in database")
+
+	return rsaKey, nil
+}
+
+func parseRSAKeyPEM(data []byte) (*rsa.PrivateKey, error) {
+	block, _ := pem.Decode(data)
+	if block == nil || block.Type != "RSA PRIVATE KEY" {
+		return nil, fmt.Errorf("failed to decode PEM block containing RSA private key")
+	}
+	return x509.ParsePKCS1PrivateKey(block.Bytes)
+}
+
+// generateAndWriteCert creates a self-signed certificate from the RSA key and writes
+// it to certPath. Returns the DER-encoded certificate bytes for the JWT x5c header.
+func generateAndWriteCert(rsaKey *rsa.PrivateKey, certPath string) ([]byte, error) {
+	template := x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject: pkix.Name{
+			Organization: []string{"ATCR"},
+			CommonName:   "ATCR Token Signing Certificate",
+		},
+		NotBefore:             time.Now(),
+		NotAfter:              time.Now().Add(10 * 365 * 24 * time.Hour),
+		KeyUsage:              x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+	}
+
+	certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &rsaKey.PublicKey, rsaKey)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create certificate: %w", err)
+	}
+
+	// Write cert to disk for distribution library
+	certPEM := pem.EncodeToMemory(&pem.Block{
+		Type:  "CERTIFICATE",
+		Bytes: certDER,
+	})
+
+	dir := filepath.Dir(certPath)
+	if err := os.MkdirAll(dir, 0700); err != nil {
+		return nil, fmt.Errorf("failed to create cert directory: %w", err)
+	}
+	if err := os.WriteFile(certPath, certPEM, 0644); err != nil {
+		return nil, fmt.Errorf("failed to write certificate: %w", err)
+	}
+
+	slog.Info("Generated JWT signing certificate", "path", certPath)
+	return certDER, nil
+}

pkg/appview/db/crypto_keys.go

@@ -0,0 +1,26 @@
+package db
+
+import "database/sql"
+
+// GetCryptoKey retrieves a key by name from the database.
+// Returns nil, nil if no key with that name exists.
+func GetCryptoKey(db DBTX, name string) ([]byte, error) {
+	var data []byte
+	err := db.QueryRow("SELECT key_data FROM crypto_keys WHERE name = ?", name).Scan(&data)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+// PutCryptoKey stores a key in the database, replacing any existing key with the same name.
+func PutCryptoKey(db DBTX, name string, data []byte) error {
+	_, err := db.Exec(
+		"INSERT INTO crypto_keys (name, key_data) VALUES (?, ?) ON CONFLICT(name) DO UPDATE SET key_data = excluded.key_data",
+		name, data,
+	)
+	return err
+}

pkg/appview/db/migrations/0013_create_crypto_keys.yaml

@@ -0,0 +1,7 @@
+description: Create crypto_keys table for storing signing keys in the database
+query: |
+  CREATE TABLE IF NOT EXISTS crypto_keys (
+      name TEXT PRIMARY KEY,
+      key_data BLOB NOT NULL,
+      created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+  );

pkg/appview/db/schema.sql

@@ -237,3 +237,9 @@ CREATE TABLE IF NOT EXISTS repo_pages (
     FOREIGN KEY(did) REFERENCES users(did) ON DELETE CASCADE
 );
 CREATE INDEX IF NOT EXISTS idx_repo_pages_did ON repo_pages(did);
+
+CREATE TABLE IF NOT EXISTS crypto_keys (
+    name TEXT PRIMARY KEY,
+    key_data BLOB NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);

pkg/appview/handlers/scan_result_test.go

@@ -84,24 +84,24 @@ func TestScanResult_WithVulnerabilities(t *testing.T) {
 
 	body := rr.Body.String()
 
-	// Should contain severity badges
-	if !strings.Contains(body, "badge-error") {
-		t.Error("Expected body to contain badge-error for critical vulnerabilities")
+	// Should contain vuln-strip severity boxes
+	if !strings.Contains(body, "vuln-box-critical") {
+		t.Error("Expected body to contain vuln-box-critical for critical vulnerabilities")
 	}
-	if !strings.Contains(body, "C:2") {
-		t.Error("Expected body to contain 'C:2' for critical count")
+	if !strings.Contains(body, `data-tip="Critical">2<`) {
+		t.Error("Expected critical count of 2")
 	}
-	if !strings.Contains(body, "badge-warning") {
-		t.Error("Expected body to contain badge-warning for high vulnerabilities")
+	if !strings.Contains(body, "vuln-box-high") {
+		t.Error("Expected body to contain vuln-box-high for high vulnerabilities")
 	}
-	if !strings.Contains(body, "H:5") {
-		t.Error("Expected body to contain 'H:5' for high count")
+	if !strings.Contains(body, `data-tip="High">5<`) {
+		t.Error("Expected high count of 5")
 	}
-	if !strings.Contains(body, "M:10") {
-		t.Error("Expected body to contain 'M:10' for medium count")
+	if !strings.Contains(body, `data-tip="Medium">10<`) {
+		t.Error("Expected medium count of 10")
 	}
-	if !strings.Contains(body, "L:3") {
-		t.Error("Expected body to contain 'L:3' for low count")
+	if !strings.Contains(body, `data-tip="Low">3<`) {
+		t.Error("Expected low count of 3")
 	}
 	// Should be clickable (has openVulnDetails)
 	if !strings.Contains(body, "openVulnDetails") {
@@ -267,8 +267,8 @@ func TestScanResult_OnlyCriticalShown(t *testing.T) {
 
 	body := rr.Body.String()
 
-	if !strings.Contains(body, "C:3") {
-		t.Error("Expected body to contain 'C:3'")
+	if !strings.Contains(body, `data-tip="Critical">3<`) {
+		t.Error("Expected critical count of 3")
 	}
 	// Zero-count badges should NOT appear
 	if strings.Contains(body, "H:0") {
@@ -346,8 +346,8 @@ func TestBatchScanResult_MultipleDigests(t *testing.T) {
 	}
 
 	// abc123 should have vulnerability badges
-	if !strings.Contains(body, "C:2") {
-		t.Error("Expected body to contain 'C:2' for abc123")
+	if !strings.Contains(body, `data-tip="Critical">2<`) {
+		t.Error("Expected critical count of 2 for abc123")
 	}
 	// def456 should have clean badge
 	if !strings.Contains(body, "Clean") {
@@ -430,7 +430,7 @@ func TestBatchScanResult_SingleDigest(t *testing.T) {
 	if !strings.Contains(body, `id="scan-badge-abc123"`) {
 		t.Error("Expected OOB span for abc123")
 	}
-	if !strings.Contains(body, "C:1") {
-		t.Error("Expected body to contain 'C:1'")
+	if !strings.Contains(body, `data-tip="Critical">1<`) {
+		t.Error("Expected critical count of 1")
 	}
 }

pkg/appview/handlers/vuln_details_test.go

@@ -17,42 +17,43 @@ import (
 
 // mockGrypeReport returns a minimal Grype JSON report
 func mockGrypeReport() string {
+	// Grype v0.107+ uses PascalCase JSON keys, and severity is nested under Metadata
 	report := map[string]any{
 		"matches": []map[string]any{
 			{
-				"vulnerability": map[string]any{
-					"id":       "CVE-2024-1234",
-					"severity": "Critical",
-					"fix":      map[string]any{"versions": []string{"1.2.4"}, "state": "fixed"},
+				"Vulnerability": map[string]any{
+					"ID":       "CVE-2024-1234",
+					"Metadata": map[string]any{"Severity": "Critical"},
+					"Fix":      map[string]any{"Versions": []string{"1.2.4"}, "State": "fixed"},
 				},
-				"artifact": map[string]any{
-					"name":    "libssl",
-					"version": "1.1.1",
-					"type":    "deb",
+				"Package": map[string]any{
+					"Name":    "libssl",
+					"Version": "1.1.1",
+					"Type":    "deb",
 				},
 			},
 			{
-				"vulnerability": map[string]any{
-					"id":       "CVE-2024-5678",
-					"severity": "Low",
-					"fix":      map[string]any{"versions": []string{}, "state": "not-fixed"},
+				"Vulnerability": map[string]any{
+					"ID":       "CVE-2024-5678",
+					"Metadata": map[string]any{"Severity": "Low"},
+					"Fix":      map[string]any{"Versions": []string{}, "State": "not-fixed"},
 				},
-				"artifact": map[string]any{
-					"name":    "zlib",
-					"version": "1.2.11",
-					"type":    "deb",
+				"Package": map[string]any{
+					"Name":    "zlib",
+					"Version": "1.2.11",
+					"Type":    "deb",
 				},
 			},
 			{
-				"vulnerability": map[string]any{
-					"id":       "GHSA-abcd-efgh-ijkl",
-					"severity": "High",
-					"fix":      map[string]any{"versions": []string{"2.0.0"}, "state": "fixed"},
+				"Vulnerability": map[string]any{
+					"ID":       "GHSA-abcd-efgh-ijkl",
+					"Metadata": map[string]any{"Severity": "High"},
+					"Fix":      map[string]any{"Versions": []string{"2.0.0"}, "State": "fixed"},
 				},
-				"artifact": map[string]any{
-					"name":    "express",
-					"version": "4.17.1",
-					"type":    "npm",
+				"Package": map[string]any{
+					"Name":    "express",
+					"Version": "4.17.1",
+					"Type":    "npm",
 				},
 			},
 		},
@@ -251,9 +252,12 @@ func TestVulnDetails_NoVulnReportBlob(t *testing.T) {
 
 	body := rr.Body.String()
 
-	// Should show summary counts
-	if !strings.Contains(body, "2 Critical") {
-		t.Error("Expected body to contain '2 Critical' summary")
+	// Should show summary counts in vuln-strip boxes
+	if !strings.Contains(body, "vuln-box-critical") {
+		t.Error("Expected body to contain vuln-box-critical in summary")
+	}
+	if !strings.Contains(body, `data-tip="Critical">2<`) {
+		t.Error("Expected critical count of 2 in summary")
 	}
 
 	// Should indicate no detailed report

pkg/appview/server.go

@@ -185,10 +185,15 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer,
 		slog.Info("TEST_MODE enabled - will use HTTP for local DID resolution")
 	}
 
+	// Load crypto keys from database (with file fallback and migration)
+	oauthKey, err := loadOAuthKey(s.Database, cfg.Server.OAuthKeyPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load OAuth key: %w", err)
+	}
+
 	// Create OAuth client app
 	desiredScopes := oauth.GetDefaultScopes(defaultHoldDID)
-	var err error
-	s.OAuthClientApp, err = oauth.NewClientApp(baseURL, s.OAuthStore, desiredScopes, cfg.Server.OAuthKeyPath, cfg.Server.ClientName)
+	s.OAuthClientApp, err = oauth.NewClientAppWithKey(baseURL, s.OAuthStore, desiredScopes, oauthKey, cfg.Server.ClientName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create OAuth client app: %w", err)
 	}
@@ -404,11 +409,12 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer,
 
 	// Create token issuer
 	if cfg.Distribution.Auth["token"] != nil {
-		s.TokenIssuer, err = s.createTokenIssuer()
+		rsaKey, certDER, err := loadJWTKeyAndCert(s.Database, cfg.Auth.KeyPath, cfg.Auth.CertPath)
 		if err != nil {
-			return nil, fmt.Errorf("failed to create token issuer: %w", err)
+			return nil, fmt.Errorf("failed to load JWT key material: %w", err)
 		}
-		slog.Info("Auth keys initialized", "path", cfg.Auth.KeyPath)
+		s.TokenIssuer = token.NewIssuerFromKey(rsaKey, certDER, cfg.Auth.ServiceName, cfg.Auth.ServiceName, cfg.Auth.TokenExpiration)
+		slog.Info("Auth keys initialized")
 	}
 
 	// Create registry app (distribution library handler)
@@ -593,15 +599,6 @@ func (s *AppViewServer) Serve() error {
 	return nil
 }
 
-// createTokenIssuer creates a token issuer for auth handlers.
-func (s *AppViewServer) createTokenIssuer() (*token.Issuer, error) {
-	return token.NewIssuer(
-		s.Config.Auth.KeyPath,
-		s.Config.Auth.ServiceName,
-		s.Config.Auth.ServiceName,
-		s.Config.Auth.TokenExpiration,
-	)
-}
 
 // DomainRoutingMiddleware enforces three-tier domain routing:
 //

pkg/auth/oauth/client.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"atcr.io/pkg/atproto"
+	"github.com/bluesky-social/indigo/atproto/atcrypto"
 	"github.com/bluesky-social/indigo/atproto/auth/oauth"
 	"github.com/bluesky-social/indigo/atproto/syntax"
 )
@@ -95,6 +96,38 @@ func NewClientApp(baseURL string, store oauth.ClientAuthStore, scopes []string,
 	return clientApp, nil
 }
 
+// NewClientAppWithKey creates an indigo OAuth ClientApp with a pre-loaded P-256 key.
+// Used by AppView when loading keys from the database instead of disk.
+// For localhost development, privateKey is ignored (public client).
+func NewClientAppWithKey(baseURL string, store oauth.ClientAuthStore, scopes []string, privateKey *atcrypto.PrivateKeyP256, clientName string) (*oauth.ClientApp, error) {
+	var config oauth.ClientConfig
+	redirectURI := RedirectURI(baseURL)
+
+	if !isLocalhost(baseURL) {
+		clientID := baseURL + "/oauth-client-metadata.json"
+		config = oauth.NewPublicConfig(clientID, redirectURI, scopes)
+
+		keyID, err := GenerateKeyID(privateKey)
+		if err != nil {
+			return nil, fmt.Errorf("failed to generate key ID: %w", err)
+		}
+
+		if err := config.SetClientSecret(privateKey, keyID); err != nil {
+			return nil, fmt.Errorf("failed to configure confidential client: %w", err)
+		}
+
+		slog.Info("Configured confidential OAuth client", "key_id", keyID)
+	} else {
+		config = oauth.NewLocalhostConfig(redirectURI, scopes)
+		slog.Info("Using public OAuth client (localhost development)")
+	}
+
+	clientApp := oauth.NewClientApp(&config, store)
+	clientApp.Dir = atproto.GetDirectory()
+
+	return clientApp, nil
+}
+
 // RedirectURI returns the OAuth redirect URI for ATCR
 func RedirectURI(baseURL string) string {
 	return baseURL + "/auth/oauth/callback"

pkg/auth/token/issuer.go

@@ -59,6 +59,19 @@ func NewIssuer(privateKeyPath, issuer, service string, expiration time.Duration)
 	}, nil
 }
 
+// NewIssuerFromKey creates a JWT issuer from pre-loaded key material.
+// certDER is the DER-encoded X.509 certificate for the x5c JWT header.
+func NewIssuerFromKey(privateKey *rsa.PrivateKey, certDER []byte, issuer, service string, expiration time.Duration) *Issuer {
+	return &Issuer{
+		privateKey:  privateKey,
+		publicKey:   &privateKey.PublicKey,
+		certificate: certDER,
+		issuer:      issuer,
+		service:     service,
+		expiration:  expiration,
+	}
+}
+
 // Issue creates and signs a new JWT token
 func (i *Issuer) Issue(subject string, access []auth.AccessEntry, authMethod string) (string, error) {
 	claims := NewClaims(subject, i.issuer, i.service, i.expiration, access, authMethod)

pkg/hold/config.go

@@ -139,6 +139,9 @@ type ServerConfig struct {
 type ScannerConfig struct {
 	// Shared secret for scanner WebSocket authentication. Empty disables scanning.
 	Secret string `yaml:"secret" comment:"Shared secret for scanner WebSocket auth. Empty disables scanning."`
+
+	// Minimum interval between re-scans of the same manifest. 0 disables proactive scanning.
+	RescanInterval time.Duration `yaml:"rescan_interval" comment:"Minimum interval between re-scans of the same manifest. When set, the hold proactively scans manifests when the scanner is idle. Default: 24h. Set to 0 to disable."`
 }
 
 // DatabaseConfig defines embedded PDS database settings
@@ -220,6 +223,7 @@ func setHoldDefaults(v *viper.Viper) {
 	v.SetDefault("gc.enabled", false)
 	// Scanner defaults
 	v.SetDefault("scanner.secret", "")
+	v.SetDefault("scanner.rescan_interval", "24h")
 
 	// Log shipper defaults
 	v.SetDefault("log_shipper.batch_size", 100)

pkg/hold/pds/scan_broadcaster.go

@@ -7,7 +7,10 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log/slog"
+	"net/http"
+	"net/url"
 	"strings"
 	"sync"
 	"time"
@@ -33,6 +36,13 @@ type ScanBroadcaster struct {
 	ackTimeout   time.Duration
 	secret       string // Shared secret for scanner authentication
 	ownsDB       bool   // true when this broadcaster opened the connection itself
+
+	// Proactive scan scheduling
+	rescanInterval   time.Duration      // Minimum interval between re-scans (0 = disabled)
+	stopCh           chan struct{}       // Signal to stop background goroutines
+	wg               sync.WaitGroup     // Wait for background goroutines to finish
+	userIdx          int                // Round-robin index through users for proactive scanning
+	predecessorCache map[string]bool    // holdDID → "is this hold's successor us?"
 }
 
 // ScanSubscriber represents a connected scanner WebSocket client
@@ -80,7 +90,7 @@ type VulnerabilitySummary struct {
 
 // NewScanBroadcaster creates a new scan job broadcaster
 // dbPath should point to a SQLite database file (e.g., "/path/to/pds/db.sqlite3")
-func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3.S3Service, holdPDS *HoldPDS) (*ScanBroadcaster, error) {
+func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3.S3Service, holdPDS *HoldPDS, rescanInterval time.Duration) (*ScanBroadcaster, error) {
 	dsn := dbPath
 	if dbPath != ":memory:" && !strings.HasPrefix(dbPath, "file:") {
 		dsn = "file:" + dbPath
@@ -107,15 +117,18 @@ func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3.
 	}
 
 	sb := &ScanBroadcaster{
-		subscribers:  make([]*ScanSubscriber, 0),
-		db:           db,
-		holdDID:      holdDID,
-		holdEndpoint: holdEndpoint,
-		s3:           s3svc,
-		pds:          holdPDS,
-		ackTimeout:   5 * time.Minute,
-		secret:       secret,
-		ownsDB:       true,
+		subscribers:      make([]*ScanSubscriber, 0),
+		db:               db,
+		holdDID:          holdDID,
+		holdEndpoint:     holdEndpoint,
+		s3:               s3svc,
+		pds:              holdPDS,
+		ackTimeout:       5 * time.Minute,
+		secret:           secret,
+		ownsDB:           true,
+		rescanInterval:   rescanInterval,
+		stopCh:           make(chan struct{}),
+		predecessorCache: make(map[string]bool),
 	}
 
 	if err := sb.initSchema(); err != nil {
@@ -124,32 +137,50 @@ func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3.
 	}
 
 	// Start re-dispatch loop for timed-out jobs
+	sb.wg.Add(1)
 	go sb.reDispatchLoop()
 
+	// Start proactive scan loop if rescan interval is configured
+	if rescanInterval > 0 {
+		sb.wg.Add(1)
+		go sb.proactiveScanLoop()
+		slog.Info("Proactive scan scheduler started", "rescanInterval", rescanInterval)
+	}
+
 	return sb, nil
 }
 
 // NewScanBroadcasterWithDB creates a scan job broadcaster using an existing *sql.DB connection.
 // The caller is responsible for the DB lifecycle.
-func NewScanBroadcasterWithDB(holdDID, holdEndpoint, secret string, db *sql.DB, s3svc *s3.S3Service, holdPDS *HoldPDS) (*ScanBroadcaster, error) {
+func NewScanBroadcasterWithDB(holdDID, holdEndpoint, secret string, db *sql.DB, s3svc *s3.S3Service, holdPDS *HoldPDS, rescanInterval time.Duration) (*ScanBroadcaster, error) {
 	sb := &ScanBroadcaster{
-		subscribers:  make([]*ScanSubscriber, 0),
-		db:           db,
-		holdDID:      holdDID,
-		holdEndpoint: holdEndpoint,
-		s3:           s3svc,
-		pds:          holdPDS,
-		ackTimeout:   5 * time.Minute,
-		secret:       secret,
-		ownsDB:       false,
+		subscribers:      make([]*ScanSubscriber, 0),
+		db:               db,
+		holdDID:          holdDID,
+		holdEndpoint:     holdEndpoint,
+		s3:               s3svc,
+		pds:              holdPDS,
+		ackTimeout:       5 * time.Minute,
+		secret:           secret,
+		ownsDB:           false,
+		rescanInterval:   rescanInterval,
+		stopCh:           make(chan struct{}),
+		predecessorCache: make(map[string]bool),
 	}
 
 	if err := sb.initSchema(); err != nil {
 		return nil, fmt.Errorf("failed to initialize scan_jobs schema: %w", err)
 	}
 
+	sb.wg.Add(1)
 	go sb.reDispatchLoop()
 
+	if rescanInterval > 0 {
+		sb.wg.Add(1)
+		go sb.proactiveScanLoop()
+		slog.Info("Proactive scan scheduler started", "rescanInterval", rescanInterval)
+	}
+
 	return sb, nil
 }
 
@@ -587,11 +618,18 @@ func (sb *ScanBroadcaster) drainPendingJobs(sub *ScanSubscriber, cursor int64) {
 
 // reDispatchLoop periodically checks for timed-out jobs and re-dispatches them
 func (sb *ScanBroadcaster) reDispatchLoop() {
+	defer sb.wg.Done()
+
 	ticker := time.NewTicker(30 * time.Second)
 	defer ticker.Stop()
 
-	for range ticker.C {
-		sb.reDispatchTimedOut()
+	for {
+		select {
+		case <-sb.stopCh:
+			return
+		case <-ticker.C:
+			sb.reDispatchTimedOut()
+		}
 	}
 }
 
@@ -649,8 +687,12 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() {
 	}
 }
 
-// Close closes the scan broadcaster's database connection
+// Close stops background goroutines and closes the scan broadcaster's database connection
 func (sb *ScanBroadcaster) Close() error {
+	if sb.stopCh != nil {
+		close(sb.stopCh)
+		sb.wg.Wait()
+	}
 	if sb.db != nil && sb.ownsDB {
 		return sb.db.Close()
 	}
@@ -667,6 +709,288 @@ func (sb *ScanBroadcaster) ValidateScannerSecret(secret string) bool {
 	return sb.secret != "" && secret == sb.secret
 }
 
+// proactiveScanLoop periodically finds manifests needing scanning and enqueues jobs.
+// It fetches manifest records from users' PDS (the source of truth) and creates scan
+// jobs for manifests that haven't been scanned recently.
+func (sb *ScanBroadcaster) proactiveScanLoop() {
+	defer sb.wg.Done()
+
+	// Wait a bit before starting to let the system settle
+	select {
+	case <-sb.stopCh:
+		return
+	case <-time.After(30 * time.Second):
+	}
+
+	ticker := time.NewTicker(60 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-sb.stopCh:
+			slog.Info("Proactive scan loop stopped")
+			return
+		case <-ticker.C:
+			sb.tryEnqueueProactiveScan()
+		}
+	}
+}
+
+// tryEnqueueProactiveScan finds the next manifest needing a scan and enqueues it.
+// Only enqueues one job per call to avoid flooding the scanner.
+func (sb *ScanBroadcaster) tryEnqueueProactiveScan() {
+	if !sb.hasConnectedScanners() {
+		return
+	}
+	if sb.hasActiveJobs() {
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+
+	// Get all users who have pushed to this hold
+	stats, err := sb.pds.ListStats(ctx)
+	if err != nil {
+		slog.Error("Proactive scan: failed to list stats", "error", err)
+		return
+	}
+
+	// Extract unique user DIDs
+	seen := make(map[string]bool)
+	var userDIDs []string
+	for _, s := range stats {
+		if !seen[s.OwnerDID] {
+			seen[s.OwnerDID] = true
+			userDIDs = append(userDIDs, s.OwnerDID)
+		}
+	}
+
+	if len(userDIDs) == 0 {
+		return
+	}
+
+	// Round-robin through users, trying each until we find work or exhaust the list
+	for attempts := 0; attempts < len(userDIDs); attempts++ {
+		idx := sb.userIdx % len(userDIDs)
+		sb.userIdx++
+		userDID := userDIDs[idx]
+
+		if sb.tryEnqueueForUser(ctx, userDID) {
+			return // Enqueued one job, done for this tick
+		}
+	}
+}
+
+// tryEnqueueForUser fetches manifests from a user's PDS and enqueues a scan for the
+// first one that needs scanning. Returns true if a job was enqueued.
+func (sb *ScanBroadcaster) tryEnqueueForUser(ctx context.Context, userDID string) bool {
+	// Resolve user DID to PDS endpoint and handle
+	did, userHandle, pdsEndpoint, err := atproto.ResolveIdentity(ctx, userDID)
+	if err != nil {
+		slog.Debug("Proactive scan: failed to resolve user identity",
+			"userDID", userDID, "error", err)
+		return false
+	}
+
+	// Fetch manifest records from user's PDS
+	client := atproto.NewClient(pdsEndpoint, did, "")
+	var cursor string
+	for {
+		records, nextCursor, err := client.ListRecordsForRepo(ctx, did, atproto.ManifestCollection, 100, cursor)
+		if err != nil {
+			slog.Debug("Proactive scan: failed to list manifest records",
+				"userDID", did, "pds", pdsEndpoint, "error", err)
+			return false
+		}
+
+		for _, record := range records {
+			var manifest atproto.ManifestRecord
+			if err := json.Unmarshal(record.Value, &manifest); err != nil {
+				slog.Debug("Proactive scan: failed to unmarshal manifest record",
+					"uri", record.URI, "error", err)
+				continue
+			}
+
+			// Check if this manifest belongs to us (directly or via successor)
+			holdDID := manifest.HoldDID
+			if holdDID == "" {
+				holdDID = manifest.HoldEndpoint // Legacy field
+			}
+			if !sb.isOurManifest(ctx, holdDID) {
+				continue
+			}
+
+			// Skip manifest lists (no layers to scan)
+			if len(manifest.Layers) == 0 {
+				continue
+			}
+
+			// Skip if config is nil (shouldn't happen for image manifests, but be safe)
+			if manifest.Config == nil {
+				continue
+			}
+
+			// Check if already scanned recently
+			if sb.isRecentlyScanned(ctx, manifest.Digest) {
+				continue
+			}
+
+			// Construct and enqueue scan job
+			configJSON, _ := json.Marshal(manifest.Config)
+			layersJSON, _ := json.Marshal(manifest.Layers)
+
+			slog.Info("Enqueuing proactive scan",
+				"manifestDigest", manifest.Digest,
+				"repository", manifest.Repository,
+				"userDID", did)
+
+			if err := sb.Enqueue(&ScanJobEvent{
+				ManifestDigest: manifest.Digest,
+				Repository:     manifest.Repository,
+				UserDID:        did,
+				UserHandle:     userHandle,
+				Tier:           "deckhand",
+				Config:         configJSON,
+				Layers:         layersJSON,
+			}); err != nil {
+				slog.Error("Proactive scan: failed to enqueue",
+					"manifest", manifest.Digest, "error", err)
+				return false
+			}
+			return true
+		}
+
+		if nextCursor == "" || len(records) == 0 {
+			break
+		}
+		cursor = nextCursor
+	}
+
+	return false
+}
+
+// isOurManifest checks if a manifest's holdDID matches this hold, either directly
+// or via successor (the manifest's hold has set us as its successor).
+func (sb *ScanBroadcaster) isOurManifest(ctx context.Context, holdDID string) bool {
+	if holdDID == "" {
+		return false
+	}
+
+	// Direct match
+	if holdDID == sb.holdDID {
+		return true
+	}
+
+	// Check predecessor cache
+	if isPredecessor, cached := sb.predecessorCache[holdDID]; cached {
+		return isPredecessor
+	}
+
+	// Fetch captain record from the other hold's PDS to check successor
+	isPredecessor := sb.checkPredecessor(ctx, holdDID)
+	sb.predecessorCache[holdDID] = isPredecessor
+	return isPredecessor
+}
+
+// checkPredecessor fetches a hold's captain record to check if its successor is us.
+func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) bool {
+	fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+
+	holdURL, err := atproto.ResolveHoldURL(fetchCtx, holdDID)
+	if err != nil {
+		slog.Debug("Proactive scan: failed to resolve predecessor hold URL",
+			"holdDID", holdDID, "error", err)
+		return false
+	}
+
+	// Fetch captain record: com.atproto.repo.getRecord
+	recordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self",
+		holdURL,
+		url.QueryEscape(holdDID),
+		url.QueryEscape(atproto.CaptainCollection),
+	)
+
+	req, err := http.NewRequestWithContext(fetchCtx, "GET", recordURL, nil)
+	if err != nil {
+		return false
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		slog.Debug("Proactive scan: failed to fetch predecessor captain record",
+			"holdDID", holdDID, "error", err)
+		return false
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return false
+	}
+
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1MB limit
+	if err != nil {
+		return false
+	}
+
+	var envelope struct {
+		Value json.RawMessage `json:"value"`
+	}
+	if err := json.Unmarshal(body, &envelope); err != nil {
+		return false
+	}
+
+	var captain atproto.CaptainRecord
+	if err := json.Unmarshal(envelope.Value, &captain); err != nil {
+		return false
+	}
+
+	if captain.Successor == sb.holdDID {
+		slog.Info("Proactive scan: discovered predecessor hold",
+			"predecessorDID", holdDID, "successor", sb.holdDID)
+		return true
+	}
+
+	return false
+}
+
+// isRecentlyScanned checks if a manifest has been scanned within the rescan interval.
+func (sb *ScanBroadcaster) isRecentlyScanned(ctx context.Context, manifestDigest string) bool {
+	_, scanRecord, err := sb.pds.GetScanRecord(ctx, manifestDigest)
+	if err != nil {
+		return false // Not scanned or error reading → needs scanning
+	}
+
+	scannedAt, err := time.Parse(time.RFC3339, scanRecord.ScannedAt)
+	if err != nil {
+		return false // Can't parse timestamp → treat as needing scan
+	}
+
+	return time.Since(scannedAt) < sb.rescanInterval
+}
+
+// hasConnectedScanners returns true if at least one scanner is connected.
+func (sb *ScanBroadcaster) hasConnectedScanners() bool {
+	sb.mu.RLock()
+	defer sb.mu.RUnlock()
+	return len(sb.subscribers) > 0
+}
+
+// hasActiveJobs returns true if there are any pending, assigned, or processing scan jobs.
+func (sb *ScanBroadcaster) hasActiveJobs() bool {
+	var count int
+	err := sb.db.QueryRow(`
+		SELECT COUNT(*) FROM scan_jobs
+		WHERE status IN ('pending', 'assigned', 'processing')
+	`).Scan(&count)
+	if err != nil {
+		slog.Error("Failed to check active scan jobs", "error", err)
+		return true // Assume busy on error
+	}
+	return count > 0
+}
+
 func generateSubscriberID() string {
 	b := make([]byte, 8)
 	_, _ = rand.Read(b)

pkg/hold/server.go

@@ -193,12 +193,13 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) {
 		// Initialize scan broadcaster if scanner secret is configured
 		if cfg.Scanner.Secret != "" {
 			holdDID := s.PDS.DID()
+			rescanInterval := cfg.Scanner.RescanInterval
 			var sb *pds.ScanBroadcaster
 			if s.holdDB != nil {
-				sb, err = pds.NewScanBroadcasterWithDB(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, s.holdDB.DB, s3Service, s.PDS)
+				sb, err = pds.NewScanBroadcasterWithDB(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, s.holdDB.DB, s3Service, s.PDS, rescanInterval)
 			} else {
 				scanDBPath := cfg.Database.Path + "/db.sqlite3"
-				sb, err = pds.NewScanBroadcaster(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, scanDBPath, s3Service, s.PDS)
+				sb, err = pds.NewScanBroadcaster(holdDID, cfg.Server.PublicURL, cfg.Scanner.Secret, scanDBPath, s3Service, s.PDS, rescanInterval)
 			}
 			if err != nil {
 				return nil, fmt.Errorf("failed to initialize scan broadcaster: %w", err)
@@ -206,7 +207,8 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) {
 			s.scanBroadcaster = sb
 			xrpcHandler.SetScanBroadcaster(sb)
 			ociHandler.SetScanBroadcaster(sb)
-			slog.Info("Scan broadcaster initialized (scanner WebSocket enabled)")
+			slog.Info("Scan broadcaster initialized (scanner WebSocket enabled)",
+				"rescanInterval", rescanInterval)
 		}
 
 		// Initialize garbage collector