Files
at-container-registry/cmd/appview/serve.go
T

802 lines
28 KiB
Go

package main
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"html/template"
"net/http"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/distribution/distribution/v3/configuration"
"github.com/distribution/distribution/v3/registry"
"github.com/distribution/distribution/v3/registry/handlers"
"github.com/spf13/cobra"
"atcr.io/pkg/appview/middleware"
"atcr.io/pkg/appview/storage"
"atcr.io/pkg/atproto"
"atcr.io/pkg/auth"
"atcr.io/pkg/auth/oauth"
"atcr.io/pkg/auth/token"
// UI components
"atcr.io/pkg/appview"
"atcr.io/pkg/appview/db"
uihandlers "atcr.io/pkg/appview/handlers"
"atcr.io/pkg/appview/holdhealth"
"atcr.io/pkg/appview/jetstream"
"atcr.io/pkg/appview/readme"
"github.com/gorilla/mux"
)
var serveCmd = &cobra.Command{
Use: "serve",
Short: "Start the ATCR registry server",
Long: `Start the ATCR registry server with authentication endpoints.
Configuration is loaded from environment variables.
See .env.appview.example for available environment variables.`,
Args: cobra.NoArgs,
RunE: serveRegistry,
}
func init() {
// Replace the default serve command with our custom one
for i, cmd := range registry.RootCmd.Commands() {
if cmd.Name() == "serve" {
registry.RootCmd.Commands()[i] = serveCmd
break
}
}
}
func serveRegistry(cmd *cobra.Command, args []string) error {
// Load configuration from environment variables
fmt.Println("Loading configuration from environment variables...")
config, err := appview.LoadConfigFromEnv()
if err != nil {
return fmt.Errorf("failed to load config from environment: %w", err)
}
fmt.Println("Configuration loaded successfully from environment")
// Initialize UI database first (required for all stores)
fmt.Println("Initializing UI database...")
uiEnabled := os.Getenv("ATCR_UI_ENABLED") != "false"
dbPath := os.Getenv("ATCR_UI_DATABASE_PATH")
if dbPath == "" {
dbPath = "/var/lib/atcr/ui.db"
}
uiDatabase, uiReadOnlyDB, uiSessionStore := db.InitializeDatabase(uiEnabled, dbPath)
if uiDatabase == nil {
return fmt.Errorf("failed to initialize UI database - required for session storage")
}
// Initialize hold health checker
fmt.Println("Initializing hold health checker...")
// Parse health check cache TTL from environment (default: 15m)
cacheTTL := 15 * time.Minute
if cacheTTLStr := os.Getenv("ATCR_HEALTH_CACHE_TTL"); cacheTTLStr != "" {
if parsed, err := time.ParseDuration(cacheTTLStr); err == nil {
cacheTTL = parsed
} else {
fmt.Printf("Warning: Invalid ATCR_HEALTH_CACHE_TTL '%s', using default 15m\n", cacheTTLStr)
}
}
healthChecker := holdhealth.NewChecker(cacheTTL)
// Initialize README cache
fmt.Println("Initializing README cache...")
readmeCacheTTL := 1 * time.Hour // Default: 1 hour
if readmeTTLStr := os.Getenv("ATCR_README_CACHE_TTL"); readmeTTLStr != "" {
if parsed, err := time.ParseDuration(readmeTTLStr); err == nil {
readmeCacheTTL = parsed
} else {
fmt.Printf("Warning: Invalid ATCR_README_CACHE_TTL '%s', using default 1h\n", readmeTTLStr)
}
}
readmeCache := readme.NewCache(uiDatabase, readmeCacheTTL)
// Start background health check worker
// Parse refresh interval from environment (default: 15m)
refreshInterval := 15 * time.Minute
if refreshIntervalStr := os.Getenv("ATCR_HEALTH_CHECK_INTERVAL"); refreshIntervalStr != "" {
if parsed, err := time.ParseDuration(refreshIntervalStr); err == nil {
refreshInterval = parsed
} else {
fmt.Printf("Warning: Invalid ATCR_HEALTH_CHECK_INTERVAL '%s', using default 15m\n", refreshIntervalStr)
}
}
startupDelay := 5 * time.Second // Wait for hold services to start (Docker compose)
dbAdapter := holdhealth.NewDBAdapter(uiDatabase)
healthWorker := holdhealth.NewWorkerWithStartupDelay(healthChecker, dbAdapter, refreshInterval, startupDelay)
// Create context for worker lifecycle management
workerCtx, workerCancel := context.WithCancel(context.Background())
defer workerCancel() // Ensure context is cancelled on all exit paths
healthWorker.Start(workerCtx)
fmt.Printf("Hold health worker started (5s startup delay, %s refresh interval, %s cache TTL)\n", refreshInterval, cacheTTL)
// Initialize OAuth components
fmt.Println("Initializing OAuth components...")
// Create OAuth session storage (SQLite-backed)
oauthStore := db.NewOAuthStore(uiDatabase)
fmt.Println("Using SQLite for OAuth session storage")
// Create device store (SQLite-backed)
deviceStore := db.NewDeviceStore(uiDatabase)
fmt.Println("Using SQLite for device storage")
// Get base URL from config or environment
baseURL := os.Getenv("ATCR_BASE_URL")
if baseURL == "" {
// If addr is just a port (e.g., ":5000"), prepend localhost
addr := config.HTTP.Addr
if addr[0] == ':' {
baseURL = fmt.Sprintf("http://127.0.0.1%s", addr)
} else {
baseURL = fmt.Sprintf("http://%s", addr)
}
}
fmt.Printf("DEBUG: Base URL for OAuth: %s\n", baseURL)
// Extract default hold DID for OAuth server and backfill worker
// This is used to create sailor profiles on first login and cache captain records
// Expected format: "did:web:hold01.atcr.io"
// To find a hold's DID, visit: https://hold01.atcr.io/.well-known/did.json
// The extraction function normalizes URLs to DIDs for consistency
defaultHoldDID := appview.ExtractDefaultHoldDID(config)
// Extract test mode from config (needed for OAuth scope configuration)
testMode := appview.ExtractTestMode(config)
if testMode {
fmt.Println("TEST_MODE enabled - will use HTTP for local DID resolution and transition:generic scope")
}
// Create OAuth app (indigo client)
oauthApp, err := oauth.NewApp(baseURL, oauthStore, defaultHoldDID, testMode)
if err != nil {
return fmt.Errorf("failed to create OAuth app: %w", err)
}
if testMode {
fmt.Println("Using OAuth scopes with transition:generic (test mode)")
} else {
fmt.Println("Using OAuth scopes with RPC scope (production mode)")
}
// Invalidate sessions with mismatched scopes on startup
// This ensures all users have the latest required scopes after deployment
desiredScopes := oauth.GetDefaultScopes(defaultHoldDID, testMode)
invalidatedCount, err := oauthStore.InvalidateSessionsWithMismatchedScopes(context.Background(), desiredScopes)
if err != nil {
fmt.Printf("Warning: Failed to invalidate sessions with mismatched scopes: %v\n", err)
} else if invalidatedCount > 0 {
fmt.Printf("Invalidated %d OAuth session(s) due to scope changes\n", invalidatedCount)
}
// Create oauth token refresher
refresher := oauth.NewRefresher(oauthApp)
// Wire up UI session store to refresher so it can invalidate UI sessions on OAuth failures
if uiSessionStore != nil {
refresher.SetUISessionStore(uiSessionStore)
}
// Set global refresher for middleware
middleware.SetGlobalRefresher(refresher)
// Set global database for pull/push metrics tracking
metricsDB := db.NewMetricsDB(uiDatabase)
middleware.SetGlobalDatabase(metricsDB)
// Create RemoteHoldAuthorizer for hold authorization with caching
holdAuthorizer := auth.NewRemoteHoldAuthorizer(uiDatabase, testMode)
middleware.SetGlobalAuthorizer(holdAuthorizer)
fmt.Println("Hold authorizer initialized with database caching")
// Set global readme cache for middleware
middleware.SetGlobalReadmeCache(readmeCache)
fmt.Println("README cache initialized for manifest push refresh")
// Initialize UI routes with OAuth app, refresher, device store, health checker, and readme cache
uiTemplates, uiRouter := initializeUIRoutes(uiDatabase, uiReadOnlyDB, uiSessionStore, oauthApp, oauthStore, refresher, baseURL, deviceStore, defaultHoldDID, healthChecker, readmeCache)
// Create OAuth server
oauthServer := oauth.NewServer(oauthApp)
// Connect server to refresher for cache invalidation
oauthServer.SetRefresher(refresher)
// Connect UI session store for web login
if uiSessionStore != nil {
oauthServer.SetUISessionStore(uiSessionStore)
}
// Register OAuth post-auth callback for AppView business logic
// This decouples the OAuth package from AppView-specific dependencies
oauthServer.SetPostAuthCallback(func(ctx context.Context, did, handle, pdsEndpoint, sessionID string) error {
fmt.Printf("DEBUG [appview/callback]: OAuth post-auth callback for DID=%s\n", did)
// Parse DID for session resume
didParsed, err := syntax.ParseDID(did)
if err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to parse DID %s: %v\n", did, err)
return nil // Non-fatal
}
// Resume OAuth session to get authenticated client
session, err := oauthApp.ResumeSession(ctx, didParsed, sessionID)
if err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to resume session for DID=%s: %v\n", did, err)
// Fallback: update user without avatar
_ = db.UpsertUser(uiDatabase, &db.User{
DID: did,
Handle: handle,
PDSEndpoint: pdsEndpoint,
Avatar: "",
LastSeen: time.Now(),
})
return nil // Non-fatal
}
// Create authenticated atproto client using the indigo session's API client
client := atproto.NewClientWithIndigoClient(pdsEndpoint, did, session.APIClient())
// Ensure sailor profile exists (creates with default hold if configured)
fmt.Printf("DEBUG [appview/callback]: Ensuring profile exists for %s (defaultHold=%s)\n", did, defaultHoldDID)
if err := storage.EnsureProfile(ctx, client, defaultHoldDID); err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to ensure profile for %s: %v\n", did, err)
// Continue anyway - profile creation is not critical for avatar fetch
} else {
fmt.Printf("DEBUG [appview/callback]: Profile ensured for %s\n", did)
}
// Fetch user's profile record from PDS (contains blob references)
profileRecord, err := client.GetProfileRecord(ctx, did)
if err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to fetch profile record for DID=%s: %v\n", did, err)
// Still update user without avatar
_ = db.UpsertUser(uiDatabase, &db.User{
DID: did,
Handle: handle,
PDSEndpoint: pdsEndpoint,
Avatar: "",
LastSeen: time.Now(),
})
return nil // Non-fatal
}
// Construct avatar URL from blob CID using imgs.blue CDN
var avatarURL string
if profileRecord.Avatar != nil && profileRecord.Avatar.Ref.Link != "" {
avatarURL = atproto.BlobCDNURL(did, profileRecord.Avatar.Ref.Link)
fmt.Printf("DEBUG [appview/callback]: Constructed avatar URL: %s\n", avatarURL)
}
// Store user with avatar in database
err = db.UpsertUser(uiDatabase, &db.User{
DID: did,
Handle: handle,
PDSEndpoint: pdsEndpoint,
Avatar: avatarURL,
LastSeen: time.Now(),
})
if err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to store user in database: %v\n", err)
return nil // Non-fatal
}
fmt.Printf("DEBUG [appview/callback]: Stored user with avatar for DID=%s\n", did)
// Migrate profile URL→DID if needed
profile, err := storage.GetProfile(ctx, client)
if err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to get profile for %s: %v\n", did, err)
return nil // Non-fatal
}
var holdDID string
if profile != nil && profile.DefaultHold != "" {
// Check if defaultHold is a URL (needs migration)
if strings.HasPrefix(profile.DefaultHold, "http://") || strings.HasPrefix(profile.DefaultHold, "https://") {
fmt.Printf("DEBUG [appview/callback]: Migrating hold URL to DID for %s: %s\n", did, profile.DefaultHold)
// Resolve URL to DID
holdDID := atproto.ResolveHoldDIDFromURL(profile.DefaultHold)
// Update profile with DID
profile.DefaultHold = holdDID
if err := storage.UpdateProfile(ctx, client, profile); err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to update profile with hold DID for %s: %v\n", did, err)
} else {
fmt.Printf("DEBUG [appview/callback]: Updated profile with hold DID: %s\n", holdDID)
}
fmt.Printf("DEBUG [oauth/server]: Attempting crew registration for %s at hold %s\n", did, holdDID)
storage.EnsureCrewMembership(ctx, client, refresher, holdDID)
} else {
// Already a DID - use it
holdDID = profile.DefaultHold
}
// Register crew regardless of migration (outside the migration block)
fmt.Printf("DEBUG [appview/callback]: Attempting crew registration for %s at hold %s\n", did, holdDID)
storage.EnsureCrewMembership(ctx, client, refresher, holdDID)
}
return nil // All errors are non-fatal, logged for debugging
})
// Initialize auth keys and create token issuer
var issuer *token.Issuer
if config.Auth["token"] != nil {
if err := initializeAuthKeys(config); err != nil {
return fmt.Errorf("failed to initialize auth keys: %w", err)
}
// Create token issuer for auth handlers
issuer, err = createTokenIssuer(config)
if err != nil {
return fmt.Errorf("failed to create token issuer: %w", err)
}
}
// Create registry app (returns http.Handler)
ctx := context.Background()
app := handlers.NewApp(ctx, config)
// Create main HTTP mux
mux := http.NewServeMux()
// Mount registry at /v2/
mux.Handle("/v2/", app)
// Mount UI routes if enabled
if uiSessionStore != nil && uiTemplates != nil && uiRouter != nil {
// Mount static files
mux.Handle("/static/", http.StripPrefix("/static/", appview.StaticHandler()))
// Mount UI routes directly at root level
mux.Handle("/", uiRouter)
fmt.Printf("UI enabled:\n")
fmt.Printf(" - Home: /\n")
fmt.Printf(" - Settings: /settings\n")
}
// Mount OAuth endpoints
mux.HandleFunc("/auth/oauth/authorize", oauthServer.ServeAuthorize)
mux.HandleFunc("/auth/oauth/callback", oauthServer.ServeCallback)
// OAuth client metadata endpoint
mux.HandleFunc("/client-metadata.json", func(w http.ResponseWriter, r *http.Request) {
config := oauthApp.GetConfig()
metadata := config.ClientMetadata()
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Access-Control-Allow-Origin", "*")
if err := json.NewEncoder(w).Encode(metadata); err != nil {
http.Error(w, "Failed to encode metadata", http.StatusInternalServerError)
}
})
// Note: Indigo handles OAuth state cleanup internally via its store
// Mount auth endpoints if enabled
if issuer != nil {
// Basic Auth token endpoint (supports device secrets and app passwords)
tokenHandler := token.NewHandler(issuer, deviceStore)
// Register token post-auth callback for profile management
// This decouples the token package from AppView-specific dependencies
tokenHandler.SetPostAuthCallback(func(ctx context.Context, did, handle, pdsEndpoint, accessToken string) error {
fmt.Printf("DEBUG [appview/callback]: Token post-auth callback for DID=%s\n", did)
// Create ATProto client with validated token
atprotoClient := atproto.NewClient(pdsEndpoint, did, accessToken)
// Ensure profile exists (will create with default hold if not exists and default is configured)
if err := storage.EnsureProfile(ctx, atprotoClient, defaultHoldDID); err != nil {
// Log error but don't fail auth - profile management is not critical
fmt.Printf("WARNING [appview/callback]: Failed to ensure profile for %s: %v\n", did, err)
} else {
fmt.Printf("DEBUG [appview/callback]: Profile ensured for %s with default hold %s\n", did, defaultHoldDID)
}
return nil // All errors are non-fatal
})
tokenHandler.RegisterRoutes(mux)
// Device authorization endpoints (public)
mux.Handle("/auth/device/code", &uihandlers.DeviceCodeHandler{
Store: deviceStore,
AppViewBaseURL: baseURL,
})
mux.Handle("/auth/device/token", &uihandlers.DeviceTokenHandler{
Store: deviceStore,
})
fmt.Printf("Auth endpoints enabled:\n")
fmt.Printf(" - Basic Auth: /auth/token (device secrets + app passwords)\n")
fmt.Printf(" - Device Auth: /auth/device/code\n")
fmt.Printf(" - Device Auth: /auth/device/token\n")
fmt.Printf(" - OAuth: /auth/oauth/authorize\n")
fmt.Printf(" - OAuth: /auth/oauth/callback\n")
fmt.Printf(" - OAuth Meta: /client-metadata.json\n")
}
// Create HTTP server
server := &http.Server{
Addr: config.HTTP.Addr,
Handler: mux,
}
// Handle graceful shutdown
stop := make(chan os.Signal, 1)
signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
// Start server in goroutine
errChan := make(chan error, 1)
go func() {
fmt.Printf("Starting registry server on %s\n", config.HTTP.Addr)
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
errChan <- err
}
}()
// Wait for shutdown signal or error
select {
case <-stop:
fmt.Println("Shutting down registry server...")
// Stop health worker first
fmt.Println("Stopping hold health worker...")
healthWorker.Stop()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := server.Shutdown(shutdownCtx); err != nil {
return fmt.Errorf("server shutdown error: %w", err)
}
case err := <-errChan:
// Stop health worker on error (workerCancel called by defer)
healthWorker.Stop()
return fmt.Errorf("server error: %w", err)
}
return nil
}
// initializeAuthKeys creates the auth keys if they don't exist
func initializeAuthKeys(config *configuration.Configuration) error {
tokenParams, ok := config.Auth["token"]
if !ok {
return nil
}
privateKeyPath := appview.GetStringParam(tokenParams, "privatekey", "/var/lib/atcr/auth/private-key.pem")
issuerName := appview.GetStringParam(tokenParams, "issuer", "atcr.io")
service := appview.GetStringParam(tokenParams, "service", "atcr.io")
expirationSecs := appview.GetIntParam(tokenParams, "expiration", 300)
// Create issuer (this will generate the key if it doesn't exist)
_, err := token.NewIssuer(
privateKeyPath,
issuerName,
service,
time.Duration(expirationSecs)*time.Second,
)
if err != nil {
return fmt.Errorf("failed to initialize token issuer: %w", err)
}
fmt.Printf("Auth keys initialized at %s\n", privateKeyPath)
return nil
}
// createTokenIssuer creates a token issuer for auth handlers
func createTokenIssuer(config *configuration.Configuration) (*token.Issuer, error) {
tokenParams, ok := config.Auth["token"]
if !ok {
return nil, fmt.Errorf("token auth not configured")
}
privateKeyPath := appview.GetStringParam(tokenParams, "privatekey", "/var/lib/atcr/auth/private-key.pem")
issuerName := appview.GetStringParam(tokenParams, "issuer", "atcr.io")
service := appview.GetStringParam(tokenParams, "service", "atcr.io")
expirationSecs := appview.GetIntParam(tokenParams, "expiration", 300)
return token.NewIssuer(
privateKeyPath,
issuerName,
service,
time.Duration(expirationSecs)*time.Second,
)
}
// initializeUIRoutes initializes the web UI routes
// database: read-write connection for auth and writes
// readOnlyDB: read-only connection for public queries (search, user pages, etc.)
// defaultHoldDID: DID of the default hold service (e.g., "did:web:hold01.atcr.io")
// healthChecker: hold endpoint health checker
func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.SessionStore, oauthApp *oauth.App, oauthStore *db.OAuthStore, refresher *oauth.Refresher, baseURL string, deviceStore *db.DeviceStore, defaultHoldDID string, healthChecker *holdhealth.Checker, readmeCache *readme.Cache) (*template.Template, *mux.Router) {
// Check if UI is enabled
uiEnabled := os.Getenv("ATCR_UI_ENABLED")
if uiEnabled == "false" {
return nil, nil
}
// Load templates
templates, err := appview.Templates()
if err != nil {
fmt.Printf("Warning: Failed to load UI templates: %v\n", err)
return nil, nil
}
// Create router
router := mux.NewRouter()
// OAuth login routes (public)
router.Handle("/auth/oauth/login", &uihandlers.LoginHandler{
Templates: templates,
}).Methods("GET")
router.Handle("/auth/oauth/login", &uihandlers.LoginSubmitHandler{}).Methods("POST")
// Public routes (with optional auth for navbar)
// SECURITY: Public pages use read-only DB
router.Handle("/", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.HomeHandler{
DB: readOnlyDB,
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
},
)).Methods("GET")
router.Handle("/api/recent-pushes", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.RecentPushesHandler{
DB: readOnlyDB,
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
HealthChecker: healthChecker,
},
)).Methods("GET")
// SECURITY: Search uses read-only DB to prevent writes and limit access to sensitive tables
router.Handle("/search", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.SearchHandler{
DB: readOnlyDB,
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
},
)).Methods("GET")
router.Handle("/api/search-results", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.SearchResultsHandler{
DB: readOnlyDB,
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
},
)).Methods("GET")
// Install page (public)
router.Handle("/install", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.InstallHandler{
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
},
)).Methods("GET")
// API route for repository stats (public, read-only)
router.Handle("/api/stats/{handle}/{repository}", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.GetStatsHandler{
DB: readOnlyDB,
Directory: oauthApp.Directory(),
},
)).Methods("GET")
// API routes for stars (require authentication)
router.Handle("/api/stars/{handle}/{repository}", middleware.RequireAuth(sessionStore, database)(
&uihandlers.StarRepositoryHandler{
DB: database, // Needs write access
Directory: oauthApp.Directory(),
Refresher: refresher,
},
)).Methods("POST")
router.Handle("/api/stars/{handle}/{repository}", middleware.RequireAuth(sessionStore, database)(
&uihandlers.UnstarRepositoryHandler{
DB: database, // Needs write access
Directory: oauthApp.Directory(),
Refresher: refresher,
},
)).Methods("DELETE")
router.Handle("/api/stars/{handle}/{repository}", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.CheckStarHandler{
DB: readOnlyDB, // Read-only check
Directory: oauthApp.Directory(),
Refresher: refresher,
},
)).Methods("GET")
// Manifest detail API endpoint
router.Handle("/api/manifests/{handle}/{repository}/{digest}", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.ManifestDetailHandler{
DB: readOnlyDB,
Directory: oauthApp.Directory(),
},
)).Methods("GET")
// Manifest health check API endpoint (HTMX polling)
router.Handle("/api/manifest-health", &uihandlers.ManifestHealthHandler{
HealthChecker: healthChecker,
}).Methods("GET")
router.Handle("/u/{handle}", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.UserPageHandler{
DB: readOnlyDB,
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
},
)).Methods("GET")
router.Handle("/r/{handle}/{repository}", middleware.OptionalAuth(sessionStore, database)(
&uihandlers.RepositoryPageHandler{
DB: readOnlyDB,
Templates: templates,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
Directory: oauthApp.Directory(),
Refresher: refresher,
HealthChecker: healthChecker,
ReadmeCache: readmeCache,
},
)).Methods("GET")
// Authenticated routes
authRouter := router.NewRoute().Subrouter()
authRouter.Use(middleware.RequireAuth(sessionStore, database))
authRouter.Handle("/settings", &uihandlers.SettingsHandler{
Templates: templates,
Refresher: refresher,
RegistryURL: uihandlers.TrimRegistryURL(baseURL),
}).Methods("GET")
authRouter.Handle("/api/profile/default-hold", &uihandlers.UpdateDefaultHoldHandler{
Refresher: refresher,
}).Methods("POST")
authRouter.Handle("/api/images/{repository}/tags/{tag}", &uihandlers.DeleteTagHandler{
DB: database,
Refresher: refresher,
}).Methods("DELETE")
authRouter.Handle("/api/images/{repository}/manifests/{digest}", &uihandlers.DeleteManifestHandler{
DB: database,
Refresher: refresher,
}).Methods("DELETE")
// Device approval page (authenticated)
authRouter.Handle("/device", &uihandlers.DeviceApprovalPageHandler{
Store: deviceStore,
SessionStore: sessionStore,
}).Methods("GET")
authRouter.Handle("/device/approve", &uihandlers.DeviceApproveHandler{
Store: deviceStore,
SessionStore: sessionStore,
}).Methods("POST")
// Device management routes
authRouter.Handle("/api/devices", &uihandlers.ListDevicesHandler{
Store: deviceStore,
SessionStore: sessionStore,
}).Methods("GET")
authRouter.Handle("/api/devices/{id}", &uihandlers.RevokeDeviceHandler{
Store: deviceStore,
SessionStore: sessionStore,
}).Methods("DELETE")
// Logout endpoint (supports both GET and POST)
// Properly revokes OAuth tokens on PDS side before clearing local session
router.Handle("/auth/logout", &uihandlers.LogoutHandler{
OAuthApp: oauthApp,
Refresher: refresher,
SessionStore: sessionStore,
OAuthStore: oauthStore,
}).Methods("GET", "POST")
// Start Jetstream worker
jetstreamURL := os.Getenv("JETSTREAM_URL")
if jetstreamURL == "" {
jetstreamURL = "wss://jetstream2.us-west.bsky.network/subscribe"
}
// Start real-time Jetstream worker with cursor tracking for reconnects
go func() {
var lastCursor int64 = 0 // Start from now on first connect
for {
worker := jetstream.NewWorker(database, jetstreamURL, lastCursor)
if err := worker.Start(context.Background()); err != nil {
// Save cursor from this connection for next reconnect
lastCursor = worker.GetLastCursor()
fmt.Printf("Jetstream: Real-time worker error: %v, reconnecting in 10s...\n", err)
time.Sleep(10 * time.Second)
}
}
}()
fmt.Println("Jetstream: Real-time worker started")
// Start backfill worker (enabled by default, set ATCR_BACKFILL_ENABLED=false to disable)
if backfillEnabled := os.Getenv("ATCR_BACKFILL_ENABLED"); backfillEnabled != "false" {
// Get relay endpoint for sync API (defaults to Bluesky's relay)
relayEndpoint := os.Getenv("ATCR_RELAY_ENDPOINT")
if relayEndpoint == "" {
relayEndpoint = "https://relay1.us-east.bsky.network"
}
// Check test mode
testMode := os.Getenv("TEST_MODE") == "true"
backfillWorker, err := jetstream.NewBackfillWorker(database, relayEndpoint, defaultHoldDID, testMode)
if err != nil {
fmt.Printf("Warning: Failed to create backfill worker: %v\n", err)
} else {
// Run initial backfill with startup delay for Docker compose
go func() {
// Wait for hold service to be ready (Docker startup race condition)
startupDelay := 5 * time.Second
fmt.Printf("Backfill: Waiting %s for services to be ready...\n", startupDelay)
time.Sleep(startupDelay)
fmt.Printf("Backfill: Starting sync-based backfill from %s...\n", relayEndpoint)
if err := backfillWorker.Start(context.Background()); err != nil {
fmt.Printf("Backfill: Finished with error: %v\n", err)
} else {
fmt.Println("Backfill: Completed successfully!")
}
}()
// Start periodic backfill scheduler
backfillInterval := os.Getenv("ATCR_BACKFILL_INTERVAL")
if backfillInterval == "" {
backfillInterval = "1h" // Default to 1 hour
}
interval, err := time.ParseDuration(backfillInterval)
if err != nil {
fmt.Printf("Warning: Invalid ATCR_BACKFILL_INTERVAL '%s', using default 1h: %v\n", backfillInterval, err)
interval = time.Hour
}
go func() {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
fmt.Printf("Backfill: Starting periodic backfill (runs every %s)...\n", interval)
if err := backfillWorker.Start(context.Background()); err != nil {
fmt.Printf("Backfill: Periodic backfill finished with error: %v\n", err)
} else {
fmt.Println("Backfill: Periodic backfill completed successfully!")
}
}
}()
fmt.Printf("Backfill: Periodic scheduler started (interval: %s)\n", interval)
}
}
return templates, router
}