mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-09-01 15:56:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

slog and refactor config in appview

Browse Source
This commit is contained in:
Evan Jarrett
2025-10-25 11:00:48 -05:00
parent 35ba417a96
commit e17600db28
20 changed files with 565 additions and 1875 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.appview.example
+1 -1
View File
@@ -66,7 +66,7 @@ ATCR_UI_ENABLED=true
# ==============================================================================
# Log level: debug, info, warn, error (default: info)
# ATCR_LOG_LEVEL=info
ATCR_LOG_LEVEL=debug
# Log formatter: text, json (default: text)
# ATCR_LOG_FORMATTER=text
.env.hold.example
+10
View File
@@ -110,3 +110,13 @@ HOLD_DATABASE_DIR=/var/lib/atcr-hold
# - Skips OAuth if records exist
#
HOLD_OWNER=did:plc:your-did-here
# ==============================================================================
# Logging Configuration
# ==============================================================================
# Log level: debug, info, warn, error (default: info)
ATCR_LOG_LEVEL=debug
# Log formatter: text, json (default: text)
# ATCR_LOG_FORMATTER=text
cmd/appview/serve.go
+102 -151
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"html/template"
"log/slog"
"net/http"
"os"
"os/signal"
@@ -14,7 +15,6 @@ import (
"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"
@@ -59,74 +59,62 @@ func init() {
}
func serveRegistry(cmd *cobra.Command, args []string) error {
// Initialize structured logging
logging.InitLogger(appview.GetLogLevel())
// Load configuration from environment variables
fmt.Println("Loading configuration from environment variables...")
config, err := appview.LoadConfigFromEnv()
cfg, 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 structured logging
logging.InitLogger(cfg.LogLevel)
slog.Info("Configuration loaded successfully from environment")
// Initialize UI database first (required for all stores)
fmt.Println("Initializing UI database...")
uiEnabled := appview.GetUIEnabled()
dbPath := appview.GetUIDatabasePath()
uiDatabase, uiReadOnlyDB, uiSessionStore := db.InitializeDatabase(uiEnabled, dbPath)
slog.Info("Initializing UI database", "path", cfg.UI.DatabasePath)
uiDatabase, uiReadOnlyDB, uiSessionStore := db.InitializeDatabase(cfg.UI.Enabled, cfg.UI.DatabasePath)
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...")
cacheTTL := appview.GetHealthCacheTTL()
healthChecker := holdhealth.NewChecker(cacheTTL)
slog.Info("Initializing hold health checker", "cache_ttl", cfg.Health.CacheTTL)
healthChecker := holdhealth.NewChecker(cfg.Health.CacheTTL)
// Initialize README cache
fmt.Println("Initializing README cache...")
readmeCacheTTL := appview.GetReadmeCacheTTL()
readmeCache := readme.NewCache(uiDatabase, readmeCacheTTL)
slog.Info("Initializing README cache", "cache_ttl", cfg.Health.ReadmeCacheTTL)
readmeCache := readme.NewCache(uiDatabase, cfg.Health.ReadmeCacheTTL)
// Start background health check worker
refreshInterval := appview.GetHealthCheckInterval()
startupDelay := 5 * time.Second // Wait for hold services to start (Docker compose)
dbAdapter := holdhealth.NewDBAdapter(uiDatabase)
healthWorker := holdhealth.NewWorkerWithStartupDelay(healthChecker, dbAdapter, refreshInterval, startupDelay)
healthWorker := holdhealth.NewWorkerWithStartupDelay(healthChecker, dbAdapter, cfg.Health.CheckInterval, 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)
slog.Info("Hold health worker started", "startup_delay", startupDelay, "refresh_interval", cfg.Health.CheckInterval, "cache_ttl", cfg.Health.CacheTTL)
// Initialize OAuth components
fmt.Println("Initializing OAuth components...")
slog.Info("Initializing OAuth components")
// Create OAuth session storage (SQLite-backed)
oauthStore := db.NewOAuthStore(uiDatabase)
fmt.Println("Using SQLite for OAuth session storage")
slog.Info("Using SQLite for OAuth session storage")
// Create device store (SQLite-backed)
deviceStore := db.NewDeviceStore(uiDatabase)
fmt.Println("Using SQLite for device storage")
slog.Info("Using SQLite for device storage")
// Get base URL from config or environment
baseURL := appview.GetBaseURL(config.HTTP.Addr)
fmt.Printf("DEBUG: Base URL for OAuth: %s\n", baseURL)
// Get base URL and default hold DID from config
baseURL := cfg.Server.BaseURL
defaultHoldDID := cfg.Server.DefaultHoldDID
testMode := cfg.Server.TestMode
// 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)
slog.Debug("Base URL for OAuth", "base_url", baseURL)
if testMode {
fmt.Println("TEST_MODE enabled - will use HTTP for local DID resolution and transition:generic scope")
slog.Info("TEST_MODE enabled - will use HTTP for local DID resolution and transition:generic scope")
}
// Create OAuth app (indigo client)
@@ -135,9 +123,9 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
return fmt.Errorf("failed to create OAuth app: %w", err)
}
if testMode {
fmt.Println("Using OAuth scopes with transition:generic (test mode)")
slog.Info("Using OAuth scopes with transition:generic (test mode)")
} else {
fmt.Println("Using OAuth scopes with RPC scope (production mode)")
slog.Info("Using OAuth scopes with RPC scope (production mode)")
}
// Invalidate sessions with mismatched scopes on startup
@@ -145,9 +133,9 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
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)
slog.Warn("Failed to invalidate sessions with mismatched scopes", "error", err)
} else if invalidatedCount > 0 {
fmt.Printf("Invalidated %d OAuth session(s) due to scope changes\n", invalidatedCount)
slog.Info("Invalidated OAuth sessions due to scope changes", "count", invalidatedCount)
}
// Create oauth token refresher
@@ -168,14 +156,17 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// Create RemoteHoldAuthorizer for hold authorization with caching
holdAuthorizer := auth.NewRemoteHoldAuthorizer(uiDatabase, testMode)
middleware.SetGlobalAuthorizer(holdAuthorizer)
fmt.Println("Hold authorizer initialized with database caching")
slog.Info("Hold authorizer initialized with database caching")
// Set global readme cache for middleware
middleware.SetGlobalReadmeCache(readmeCache)
fmt.Println("README cache initialized for manifest push refresh")
slog.Info("README cache initialized for manifest push refresh")
// Initialize Jetstream workers (background services before HTTP routes)
initializeJetstream(uiDatabase, &cfg.Jetstream, defaultHoldDID, testMode)
// 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)
uiTemplates, uiRouter := initializeUIRoutes(cfg.UI.Enabled, uiDatabase, uiReadOnlyDB, uiSessionStore, oauthApp, oauthStore, refresher, baseURL, deviceStore, healthChecker, readmeCache)
// Create OAuth server
oauthServer := oauth.NewServer(oauthApp)
@@ -189,19 +180,19 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// 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)
slog.Debug("OAuth post-auth callback", "component", "appview/callback", "did", 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)
slog.Warn("Failed to parse DID", "component", "appview/callback", "did", did, "error", 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)
slog.Warn("Failed to resume session", "component", "appview/callback", "did", did, "error", err)
// Fallback: update user without avatar
_ = db.UpsertUser(uiDatabase, &db.User{
DID: did,
@@ -217,18 +208,18 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
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)
slog.Debug("Ensuring profile exists", "component", "appview/callback", "did", did, "default_hold_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)
slog.Warn("Failed to ensure profile", "component", "appview/callback", "did", did, "error", err)
// Continue anyway - profile creation is not critical for avatar fetch
} else {
fmt.Printf("DEBUG [appview/callback]: Profile ensured for %s\n", did)
slog.Debug("Profile ensured", "component", "appview/callback", "did", 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)
slog.Warn("Failed to fetch profile record", "component", "appview/callback", "did", did, "error", err)
// Still update user without avatar
_ = db.UpsertUser(uiDatabase, &db.User{
DID: did,
@@ -244,7 +235,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
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)
slog.Debug("Constructed avatar URL", "component", "appview/callback", "avatar_url", avatarURL)
}
// Store user with avatar in database
@@ -256,16 +247,16 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
LastSeen: time.Now(),
})
if err != nil {
fmt.Printf("WARNING [appview/callback]: Failed to store user in database: %v\n", err)
slog.Warn("Failed to store user in database", "component", "appview/callback", "error", err)
return nil // Non-fatal
}
fmt.Printf("DEBUG [appview/callback]: Stored user with avatar for DID=%s\n", did)
slog.Debug("Stored user with avatar", "component", "appview/callback", "did", 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)
slog.Warn("Failed to get profile", "component", "appview/callback", "did", did, "error", err)
return nil // Non-fatal
}
@@ -273,7 +264,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
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)
slog.Debug("Migrating hold URL to DID", "component", "appview/callback", "did", did, "hold_url", profile.DefaultHold)
// Resolve URL to DID
holdDID := atproto.ResolveHoldDIDFromURL(profile.DefaultHold)
@@ -281,18 +272,18 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// 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)
slog.Warn("Failed to update profile with hold DID", "component", "appview/callback", "did", did, "error", err)
} else {
fmt.Printf("DEBUG [appview/callback]: Updated profile with hold DID: %s\n", holdDID)
slog.Debug("Updated profile with hold DID", "component", "appview/callback", "hold_did", holdDID)
}
fmt.Printf("DEBUG [oauth/server]: Attempting crew registration for %s at hold %s\n", did, holdDID)
slog.Debug("Attempting crew registration", "component", "oauth/server", "did", did, "hold_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)
slog.Debug("Attempting crew registration", "component", "appview/callback", "did", did, "hold_did", holdDID)
storage.EnsureCrewMembership(ctx, client, refresher, holdDID)
}
@@ -300,23 +291,21 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
return nil // All errors are non-fatal, logged for debugging
})
// Initialize auth keys and create token issuer
// Create token issuer (also initializes auth keys if needed)
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 cfg.Distribution.Auth["token"] != nil {
issuer, err = createTokenIssuer(cfg)
if err != nil {
return fmt.Errorf("failed to create token issuer: %w", err)
}
// Log successful initialization
slog.Info("Auth keys initialized", "path", cfg.Auth.KeyPath)
}
// Create registry app (returns http.Handler)
ctx := context.Background()
app := handlers.NewApp(ctx, config)
app := handlers.NewApp(ctx, cfg.Distribution)
// Create main HTTP mux
mux := http.NewServeMux()
@@ -332,9 +321,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// Mount UI routes directly at root level
mux.Handle("/", uiRouter)
fmt.Printf("UI enabled:\n")
fmt.Printf(" - Home: /\n")
fmt.Printf(" - Settings: /settings\n")
slog.Info("UI enabled", "home", "/", "settings", "/settings")
}
// Mount OAuth endpoints
@@ -363,7 +350,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// 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)
slog.Debug("Token post-auth callback", "component", "appview/callback", "did", did)
// Create ATProto client with validated token
atprotoClient := atproto.NewClient(pdsEndpoint, did, accessToken)
@@ -371,9 +358,9 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// 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)
slog.Warn("Failed to ensure profile", "component", "appview/callback", "did", did, "error", err)
} else {
fmt.Printf("DEBUG [appview/callback]: Profile ensured for %s with default hold %s\n", did, defaultHoldDID)
slog.Debug("Profile ensured with default hold", "component", "appview/callback", "did", did, "default_hold_did", defaultHoldDID)
}
return nil // All errors are non-fatal
@@ -390,18 +377,18 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
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")
slog.Info("Auth endpoints enabled",
"basic_auth", "/auth/token",
"device_code", "/auth/device/code",
"device_token", "/auth/device/token",
"oauth_authorize", "/auth/oauth/authorize",
"oauth_callback", "/auth/oauth/callback",
"oauth_metadata", "/client-metadata.json")
}
// Create HTTP server
server := &http.Server{
Addr: config.HTTP.Addr,
Addr: cfg.Server.Addr,
Handler: mux,
}
@@ -412,7 +399,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// Start server in goroutine
errChan := make(chan error, 1)
go func() {
fmt.Printf("Starting registry server on %s\n", config.HTTP.Addr)
slog.Info("Starting registry server", "addr", cfg.Server.Addr)
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
errChan <- err
}
@@ -421,10 +408,10 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
// Wait for shutdown signal or error
select {
case <-stop:
fmt.Println("Shutting down registry server...")
slog.Info("Shutting down registry server")
// Stop health worker first
fmt.Println("Stopping hold health worker...")
slog.Info("Stopping hold health worker")
healthWorker.Stop()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@@ -442,68 +429,32 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
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)
func createTokenIssuer(cfg *appview.Config) (*token.Issuer, error) {
return token.NewIssuer(
privateKeyPath,
issuerName,
service,
time.Duration(expirationSecs)*time.Second,
cfg.Auth.KeyPath,
cfg.Auth.ServiceName, // issuer
cfg.Auth.ServiceName, // service
cfg.Auth.TokenExpiration,
)
}
// initializeUIRoutes initializes the web UI routes
// uiEnabled: whether UI is enabled (from Config.UI.Enabled)
// 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) {
// readmeCache: README cache for repository pages
func initializeUIRoutes(uiEnabled bool, database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.SessionStore, oauthApp *oauth.App, oauthStore *db.OAuthStore, refresher *oauth.Refresher, baseURL string, deviceStore *db.DeviceStore, healthChecker *holdhealth.Checker, readmeCache *readme.Cache) (*template.Template, *mux.Router) {
// Check if UI is enabled
if !appview.GetUIEnabled() {
if !uiEnabled {
return nil, nil
}
// Load templates
templates, err := appview.Templates()
if err != nil {
fmt.Printf("Warning: Failed to load UI templates: %v\n", err)
slog.Warn("Failed to load UI templates", "error", err)
return nil, nil
}
@@ -682,8 +633,13 @@ func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.S
OAuthStore: oauthStore,
}).Methods("GET", "POST")
return templates, router
}
// initializeJetstream initializes the Jetstream workers for real-time events and backfill
func initializeJetstream(database *sql.DB, jetstreamCfg *appview.JetstreamConfig, defaultHoldDID string, testMode bool) {
// Start Jetstream worker
jetstreamURL := appview.GetJetstreamURL()
jetstreamURL := jetstreamCfg.URL
// Start real-time Jetstream worker with cursor tracking for reconnects
go func() {
@@ -693,59 +649,54 @@ func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.S
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)
slog.Warn("Jetstream real-time worker error, reconnecting", "component", "jetstream", "error", err, "reconnect_delay", "10s")
time.Sleep(10 * time.Second)
}
}
}()
fmt.Println("Jetstream: Real-time worker started")
slog.Info("Jetstream real-time worker started", "component", "jetstream")
// Start backfill worker (enabled by default, set ATCR_BACKFILL_ENABLED=false to disable)
if appview.GetBackfillEnabled() {
if jetstreamCfg.BackfillEnabled {
// Get relay endpoint for sync API (defaults to Bluesky's relay)
relayEndpoint := appview.GetRelayEndpoint()
// Check test mode
testMode := appview.GetTestMode()
relayEndpoint := jetstreamCfg.RelayEndpoint
backfillWorker, err := jetstream.NewBackfillWorker(database, relayEndpoint, defaultHoldDID, testMode)
if err != nil {
fmt.Printf("Warning: Failed to create backfill worker: %v\n", err)
slog.Warn("Failed to create backfill worker", "component", "jetstream/backfill", "error", 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)
slog.Info("Waiting for services to be ready", "component", "jetstream/backfill", "startup_delay", startupDelay)
time.Sleep(startupDelay)
fmt.Printf("Backfill: Starting sync-based backfill from %s...\n", relayEndpoint)
slog.Info("Starting sync-based backfill", "component", "jetstream/backfill", "relay_endpoint", relayEndpoint)
if err := backfillWorker.Start(context.Background()); err != nil {
fmt.Printf("Backfill: Finished with error: %v\n", err)
slog.Warn("Backfill finished with error", "component", "jetstream/backfill", "error", err)
} else {
fmt.Println("Backfill: Completed successfully!")
slog.Info("Backfill completed successfully", "component", "jetstream/backfill")
}
}()
// Start periodic backfill scheduler
interval := appview.GetBackfillInterval()
interval := jetstreamCfg.BackfillInterval
go func() {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
fmt.Printf("Backfill: Starting periodic backfill (runs every %s)...\n", interval)
slog.Info("Starting periodic backfill", "component", "jetstream/backfill", "interval", interval)
if err := backfillWorker.Start(context.Background()); err != nil {
fmt.Printf("Backfill: Periodic backfill finished with error: %v\n", err)
slog.Warn("Periodic backfill finished with error", "component", "jetstream/backfill", "error", err)
} else {
fmt.Println("Backfill: Periodic backfill completed successfully!")
slog.Info("Periodic backfill completed successfully", "component", "jetstream/backfill")
}
}
}()
fmt.Printf("Backfill: Periodic scheduler started (interval: %s)\n", interval)
slog.Info("Periodic backfill scheduler started", "component", "jetstream/backfill", "interval", interval)
}
}
return templates, router
}
deploy/.env.prod.template
+1 -1
View File
@@ -161,7 +161,7 @@ ATCR_UI_ENABLED=true
# Log level: debug, info, warn, error
# Default: info
ATCR_LOG_LEVEL=info
ATCR_LOG_LEVEL=debug
# Log formatter: text, json
# Default: text
deploy/docker-compose.prod.yml
+4
View File
@@ -114,6 +114,10 @@ services:
S3_ENDPOINT: ${S3_ENDPOINT:-}
S3_REGION_ENDPOINT: ${S3_REGION_ENDPOINT:-}
# Logging
ATCR_LOG_LEVEL: ${ATCR_LOG_LEVEL:-debug}
ATCR_LOG_FORMATTER: ${ATCR_LOG_FORMATTER:-text}
# Optional: Filesystem storage (comment out S3 vars above)
# STORAGE_DRIVER: filesystem
# STORAGE_ROOT_DIR: /var/lib/atcr/hold
docker-compose.yml
+3 -1
View File
@@ -20,7 +20,7 @@ services:
# Test mode - fallback to default hold when user's hold is unreachable
TEST_MODE: true
# Logging
ATCR_LOG_LEVEL: info
ATCR_LOG_LEVEL: debug
volumes:
# Auth keys (JWT signing keys)
# - atcr-auth:/var/lib/atcr/auth
@@ -50,6 +50,8 @@ services:
# STORAGE_ROOT_DIR: /var/lib/atcr/hold
TEST_MODE: true
# DISABLE_PRESIGNED_URLS: true
# Logging
ATCR_LOG_LEVEL: debug
# Storage config comes from env_file (STORAGE_DRIVER, AWS_*, S3_*)
build:
context: .
pkg/appview/config.go
+205 -280
View File
@@ -17,92 +17,230 @@ import (
"github.com/distribution/distribution/v3/configuration"
)
// LoadConfigFromEnv builds a complete configuration from environment variables
// This follows the same pattern as the hold service (no config files, only env vars)
func LoadConfigFromEnv() (*configuration.Configuration, error) {
config := &configuration.Configuration{}
// Version
config.Version = configuration.MajorMinorVersion(0, 1)
// Logging
config.Log = buildLogConfig()
// HTTP server
httpConfig, err := buildHTTPConfig()
if err != nil {
return nil, fmt.Errorf("failed to build HTTP config: %w", err)
}
config.HTTP = httpConfig
// Storage (fake in-memory placeholder - all real storage is proxied)
config.Storage = buildStorageConfig()
// Get base URL for error messages and auth config
baseURL := GetBaseURL(httpConfig.Addr)
// Middleware (ATProto resolver)
defaultHoldDID := os.Getenv("ATCR_DEFAULT_HOLD_DID")
if defaultHoldDID == "" {
return nil, fmt.Errorf("ATCR_DEFAULT_HOLD_DID is required")
}
config.Middleware = buildMiddlewareConfig(defaultHoldDID, baseURL)
// Auth
authConfig, err := buildAuthConfig(baseURL)
if err != nil {
return nil, fmt.Errorf("failed to build auth config: %w", err)
}
config.Auth = authConfig
// Health checks
config.Health = buildHealthConfig()
return config, nil
// Config represents the AppView service configuration
type Config struct {
Version string `yaml:"version"`
LogLevel string `yaml:"log_level"`
Server ServerConfig `yaml:"server"`
UI UIConfig `yaml:"ui"`
Health HealthConfig `yaml:"health"`
Jetstream JetstreamConfig `yaml:"jetstream"`
Auth AuthConfig `yaml:"auth"`
Distribution *configuration.Configuration `yaml:"-"` // Wrapped distribution config for compatibility
}
// buildLogConfig creates logging configuration from environment variables
func buildLogConfig() configuration.Log {
level := GetEnvOrDefault("ATCR_LOG_LEVEL", "info")
formatter := GetEnvOrDefault("ATCR_LOG_FORMATTER", "text")
// ServerConfig defines server settings
type ServerConfig struct {
// Addr is the HTTP listen address (from env: ATCR_HTTP_ADDR, default: ":5000")
Addr string `yaml:"addr"`
return configuration.Log{
Level: configuration.Loglevel(level),
Formatter: formatter,
// BaseURL is the public URL for OAuth/JWT realm (from env: ATCR_BASE_URL)
// Auto-detected from Addr if not set
BaseURL string `yaml:"base_url"`
// DefaultHoldDID is the default hold DID for blob storage (from env: ATCR_DEFAULT_HOLD_DID)
// REQUIRED - e.g., "did:web:hold01.atcr.io"
DefaultHoldDID string `yaml:"default_hold_did"`
// TestMode enables HTTP for local DID resolution and transition:generic scope (from env: TEST_MODE)
TestMode bool `yaml:"test_mode"`
// DebugAddr is the debug/pprof HTTP listen address (from env: ATCR_DEBUG_ADDR, default: ":5001")
DebugAddr string `yaml:"debug_addr"`
}
// UIConfig defines web UI settings
type UIConfig struct {
// Enabled controls whether the web UI is enabled (from env: ATCR_UI_ENABLED, default: true)
Enabled bool `yaml:"enabled"`
// DatabasePath is the path to the UI SQLite database (from env: ATCR_UI_DATABASE_PATH, default: "/var/lib/atcr/ui.db")
DatabasePath string `yaml:"database_path"`
}
// HealthConfig defines health check and cache settings
type HealthConfig struct {
// CacheTTL is the hold health check cache TTL (from env: ATCR_HEALTH_CACHE_TTL, default: 15m)
CacheTTL time.Duration `yaml:"cache_ttl"`
// CheckInterval is the hold health check refresh interval (from env: ATCR_HEALTH_CHECK_INTERVAL, default: 15m)
CheckInterval time.Duration `yaml:"check_interval"`
// ReadmeCacheTTL is the README cache TTL (from env: ATCR_README_CACHE_TTL, default: 1h)
ReadmeCacheTTL time.Duration `yaml:"readme_cache_ttl"`
}
// JetstreamConfig defines ATProto Jetstream settings
type JetstreamConfig struct {
// URL is the Jetstream WebSocket URL (from env: JETSTREAM_URL, default: wss://jetstream2.us-west.bsky.network/subscribe)
URL string `yaml:"url"`
// BackfillEnabled controls whether backfill is enabled (from env: ATCR_BACKFILL_ENABLED, default: true)
BackfillEnabled bool `yaml:"backfill_enabled"`
// BackfillInterval is the backfill interval (from env: ATCR_BACKFILL_INTERVAL, default: 1h)
BackfillInterval time.Duration `yaml:"backfill_interval"`
// RelayEndpoint is the relay endpoint for sync API (from env: ATCR_RELAY_ENDPOINT, default: https://relay1.us-east.bsky.network)
RelayEndpoint string `yaml:"relay_endpoint"`
}
// AuthConfig defines authentication settings
type AuthConfig struct {
// KeyPath is the JWT signing key path (from env: ATCR_AUTH_KEY_PATH, default: "/var/lib/atcr/auth/private-key.pem")
KeyPath string `yaml:"key_path"`
// CertPath is the JWT certificate path (from env: ATCR_AUTH_CERT_PATH, default: "/var/lib/atcr/auth/private-key.crt")
CertPath string `yaml:"cert_path"`
// TokenExpiration is the JWT expiration duration (from env: ATCR_TOKEN_EXPIRATION, default: 300s)
TokenExpiration time.Duration `yaml:"token_expiration"`
// ServiceName is the service name used for JWT issuer and service fields
// Derived from ATCR_SERVICE_NAME env var or extracted from base URL (e.g., "atcr.io")
ServiceName string `yaml:"service_name"`
}
// LoadConfigFromEnv builds a complete configuration from environment variables
// This follows the same pattern as the hold service (no config files, only env vars)
func LoadConfigFromEnv() (*Config, error) {
cfg := &Config{
Version: "0.1",
}
// Logging configuration
cfg.LogLevel = getEnvOrDefault("ATCR_LOG_LEVEL", "info")
// Server configuration
cfg.Server.Addr = getEnvOrDefault("ATCR_HTTP_ADDR", ":5000")
cfg.Server.DebugAddr = getEnvOrDefault("ATCR_DEBUG_ADDR", ":5001")
cfg.Server.DefaultHoldDID = os.Getenv("ATCR_DEFAULT_HOLD_DID")
if cfg.Server.DefaultHoldDID == "" {
return nil, fmt.Errorf("ATCR_DEFAULT_HOLD_DID is required")
}
cfg.Server.TestMode = os.Getenv("TEST_MODE") == "true"
// Auto-detect base URL if not explicitly set
cfg.Server.BaseURL = os.Getenv("ATCR_BASE_URL")
if cfg.Server.BaseURL == "" {
cfg.Server.BaseURL = autoDetectBaseURL(cfg.Server.Addr)
}
// UI configuration
cfg.UI.Enabled = os.Getenv("ATCR_UI_ENABLED") != "false"
cfg.UI.DatabasePath = getEnvOrDefault("ATCR_UI_DATABASE_PATH", "/var/lib/atcr/ui.db")
// Health and cache configuration
cfg.Health.CacheTTL = getDurationOrDefault("ATCR_HEALTH_CACHE_TTL", 15*time.Minute)
cfg.Health.CheckInterval = getDurationOrDefault("ATCR_HEALTH_CHECK_INTERVAL", 15*time.Minute)
cfg.Health.ReadmeCacheTTL = getDurationOrDefault("ATCR_README_CACHE_TTL", 1*time.Hour)
// Jetstream configuration
cfg.Jetstream.URL = getEnvOrDefault("JETSTREAM_URL", "wss://jetstream2.us-west.bsky.network/subscribe")
cfg.Jetstream.BackfillEnabled = os.Getenv("ATCR_BACKFILL_ENABLED") != "false"
cfg.Jetstream.BackfillInterval = getDurationOrDefault("ATCR_BACKFILL_INTERVAL", 1*time.Hour)
cfg.Jetstream.RelayEndpoint = getEnvOrDefault("ATCR_RELAY_ENDPOINT", "https://relay1.us-east.bsky.network")
// Auth configuration
cfg.Auth.KeyPath = getEnvOrDefault("ATCR_AUTH_KEY_PATH", "/var/lib/atcr/auth/private-key.pem")
cfg.Auth.CertPath = getEnvOrDefault("ATCR_AUTH_CERT_PATH", "/var/lib/atcr/auth/private-key.crt")
// Parse token expiration (default: 300 seconds = 5 minutes)
expirationStr := getEnvOrDefault("ATCR_TOKEN_EXPIRATION", "300")
expirationSecs, err := strconv.Atoi(expirationStr)
if err != nil {
return nil, fmt.Errorf("invalid ATCR_TOKEN_EXPIRATION: %w", err)
}
cfg.Auth.TokenExpiration = time.Duration(expirationSecs) * time.Second
// Derive service name from base URL or env var (used for JWT issuer and service)
cfg.Auth.ServiceName = getServiceName(cfg.Server.BaseURL)
// Build distribution configuration for compatibility with distribution library
distConfig, err := buildDistributionConfig(cfg)
if err != nil {
return nil, fmt.Errorf("failed to build distribution config: %w", err)
}
cfg.Distribution = distConfig
return cfg, nil
}
// buildDistributionConfig creates a distribution Configuration from our Config
// This maintains compatibility with the distribution library
func buildDistributionConfig(cfg *Config) (*configuration.Configuration, error) {
distConfig := &configuration.Configuration{}
// Version
distConfig.Version = configuration.MajorMinorVersion(0, 1)
// Logging
distConfig.Log = configuration.Log{
Level: configuration.Loglevel(cfg.LogLevel),
Formatter: getEnvOrDefault("ATCR_LOG_FORMATTER", "text"),
Fields: map[string]any{
"service": "atcr-appview",
},
}
}
// buildHTTPConfig creates HTTP server configuration from environment variables
func buildHTTPConfig() (configuration.HTTP, error) {
addr := GetEnvOrDefault("ATCR_HTTP_ADDR", ":5000")
debugAddr := GetEnvOrDefault("ATCR_DEBUG_ADDR", ":5001")
// HTTP secret - only needed for multipart uploads in distribution's storage driver
// Since AppView is stateless and routes all storage through middleware, this isn't
// actually used, but we generate a random secret for defense in depth
// HTTP server
httpSecret := os.Getenv("REGISTRY_HTTP_SECRET")
if httpSecret == "" {
// Generate a random 32-byte secret
randomBytes := make([]byte, 32)
if _, err := rand.Read(randomBytes); err != nil {
return configuration.HTTP{}, fmt.Errorf("failed to generate random secret: %w", err)
return nil, fmt.Errorf("failed to generate random secret: %w", err)
}
httpSecret = hex.EncodeToString(randomBytes)
}
return configuration.HTTP{
Addr: addr,
distConfig.HTTP = configuration.HTTP{
Addr: cfg.Server.Addr,
Secret: httpSecret,
Headers: map[string][]string{
"X-Content-Type-Options": {"nosniff"},
},
Debug: configuration.Debug{
Addr: debugAddr,
Addr: cfg.Server.DebugAddr,
},
}, nil
}
// Storage (fake in-memory placeholder - all real storage is proxied)
distConfig.Storage = buildStorageConfig()
// Middleware (ATProto resolver)
distConfig.Middleware = buildMiddlewareConfig(cfg.Server.DefaultHoldDID, cfg.Server.BaseURL)
// Auth (use values from cfg.Auth)
realm := cfg.Server.BaseURL + "/auth/token"
distConfig.Auth = configuration.Auth{
"token": configuration.Parameters{
"realm": realm,
"service": cfg.Auth.ServiceName,
"issuer": cfg.Auth.ServiceName,
"rootcertbundle": cfg.Auth.CertPath,
"privatekey": cfg.Auth.KeyPath,
"expiration": int(cfg.Auth.TokenExpiration.Seconds()),
},
}
// Health checks
distConfig.Health = buildHealthConfig()
return distConfig, nil
}
// autoDetectBaseURL determines the base URL for the service from the HTTP address
func autoDetectBaseURL(httpAddr string) string {
// Auto-detect from HTTP addr
if httpAddr[0] == ':' {
// Just a port, assume localhost
return fmt.Sprintf("http://127.0.0.1%s", httpAddr)
}
// Full address provided
return fmt.Sprintf("http://%s", httpAddr)
}
// buildStorageConfig creates a fake in-memory storage config
@@ -148,37 +286,6 @@ func buildMiddlewareConfig(defaultHoldDID string, baseURL string) map[string][]c
}
}
// buildAuthConfig creates authentication configuration from environment variables
func buildAuthConfig(baseURL string) (configuration.Auth, error) {
// Token configuration
privateKeyPath := GetEnvOrDefault("ATCR_AUTH_KEY_PATH", "/var/lib/atcr/auth/private-key.pem")
certPath := GetEnvOrDefault("ATCR_AUTH_CERT_PATH", "/var/lib/atcr/auth/private-key.crt")
// Token expiration in seconds (default: 5 minutes)
expirationStr := GetEnvOrDefault("ATCR_TOKEN_EXPIRATION", "300")
expiration, err := strconv.Atoi(expirationStr)
if err != nil {
return configuration.Auth{}, fmt.Errorf("invalid ATCR_TOKEN_EXPIRATION: %w", err)
}
// Auto-derive service name from base URL or use env var
serviceName := getServiceName(baseURL)
// Auto-derive realm from base URL
realm := baseURL + "/auth/token"
return configuration.Auth{
"token": configuration.Parameters{
"realm": realm,
"service": serviceName,
"issuer": serviceName,
"rootcertbundle": certPath,
"privatekey": privateKeyPath,
"expiration": expiration,
},
}, nil
}
// buildHealthConfig creates health check configuration
func buildHealthConfig() configuration.Health {
return configuration.Health{
@@ -190,24 +297,6 @@ func buildHealthConfig() configuration.Health {
}
}
// GetBaseURL determines the base URL for the service
// Priority: ATCR_BASE_URL env var, then derived from HTTP addr
func GetBaseURL(httpAddr string) string {
baseURL := os.Getenv("ATCR_BASE_URL")
if baseURL != "" {
return baseURL
}
// Auto-detect from HTTP addr
if httpAddr[0] == ':' {
// Just a port, assume localhost
return fmt.Sprintf("http://127.0.0.1%s", httpAddr)
}
// Full address provided
return fmt.Sprintf("http://%s", httpAddr)
}
// getServiceName extracts service name from base URL or uses env var
func getServiceName(baseURL string) string {
// Check env var first
@@ -232,98 +321,17 @@ func getServiceName(baseURL string) string {
return "atcr.io"
}
// GetEnvOrDefault gets an environment variable or returns a default value
func GetEnvOrDefault(key, defaultValue string) string {
// getEnvOrDefault gets an environment variable or returns a default value
func getEnvOrDefault(key, defaultValue string) string {
if val := os.Getenv(key); val != "" {
return val
}
return defaultValue
}
// GetLogLevel returns the configured log level from environment
// Centralizes ATCR_LOG_LEVEL env var reading
func GetLogLevel() string {
return GetEnvOrDefault("ATCR_LOG_LEVEL", "info")
}
// GetStringParam extracts a string parameter from configuration.Parameters
func GetStringParam(params configuration.Parameters, key, defaultValue string) string {
if v, ok := params[key]; ok {
if s, ok := v.(string); ok {
return s
}
}
return defaultValue
}
// GetIntParam extracts an int parameter from configuration.Parameters
func GetIntParam(params configuration.Parameters, key string, defaultValue int) int {
if v, ok := params[key]; ok {
if i, ok := v.(int); ok {
return i
}
}
return defaultValue
}
// ExtractDefaultHoldDID extracts the default hold DID from middleware config
// Returns a DID (e.g., "did:web:hold01.atcr.io")
// To find a hold's DID, visit: https://hold-url/.well-known/did.json
func ExtractDefaultHoldDID(config *configuration.Configuration) string {
// Navigate through: middleware.registry[].options.default_hold_did
registryMiddleware, ok := config.Middleware["registry"]
if !ok {
return ""
}
// Find atproto-resolver middleware
for _, mw := range registryMiddleware {
// Check if this is the atproto-resolver
if mw.Name != "atproto-resolver" {
continue
}
// Extract options - options is configuration.Parameters which is map[string]any
if mw.Options != nil {
if holdDID, ok := mw.Options["default_hold_did"].(string); ok {
return holdDID
}
}
}
return ""
}
// ExtractTestMode extracts the test_mode flag from middleware config
// Returns true if TEST_MODE=true, false otherwise
func ExtractTestMode(config *configuration.Configuration) bool {
// Navigate through: middleware.registry[].options.test_mode
registryMiddleware, ok := config.Middleware["registry"]
if !ok {
return false
}
// Find atproto-resolver middleware
for _, mw := range registryMiddleware {
// Check if this is the atproto-resolver
if mw.Name != "atproto-resolver" {
continue
}
// Extract options - options is configuration.Parameters which is map[string]any
if mw.Options != nil {
if testMode, ok := mw.Options["test_mode"].(bool); ok {
return testMode
}
}
}
return false
}
// GetDurationOrDefault parses a duration from environment variable or returns default
// getDurationOrDefault parses a duration from environment variable or returns default
// Logs a warning if parsing fails
func GetDurationOrDefault(envKey string, defaultValue time.Duration) time.Duration {
func getDurationOrDefault(envKey string, defaultValue time.Duration) time.Duration {
envVal := os.Getenv(envKey)
if envVal == "" {
return defaultValue
@@ -337,86 +345,3 @@ func GetDurationOrDefault(envKey string, defaultValue time.Duration) time.Durati
return parsed
}
// GetBoolOrDefault returns a boolean from environment variable or returns default
// Treats "false" as false, everything else (including empty) as the default value
func GetBoolOrDefault(envKey string, defaultValue bool) bool {
envVal := os.Getenv(envKey)
if envVal == "" {
return defaultValue
}
// Explicit false check
if envVal == "false" {
return false
}
// Explicit true check
if envVal == "true" {
return true
}
// For any other value, return default
return defaultValue
}
// UI Configuration
// GetUIEnabled returns whether the UI is enabled (default: true)
func GetUIEnabled() bool {
// UI is enabled unless explicitly set to "false"
return os.Getenv("ATCR_UI_ENABLED") != "false"
}
// GetUIDatabasePath returns the path to the UI database (default: /var/lib/atcr/ui.db)
func GetUIDatabasePath() string {
return GetEnvOrDefault("ATCR_UI_DATABASE_PATH", "/var/lib/atcr/ui.db")
}
// Health & Cache Configuration
// GetHealthCacheTTL returns the hold health check cache TTL (default: 15m)
func GetHealthCacheTTL() time.Duration {
return GetDurationOrDefault("ATCR_HEALTH_CACHE_TTL", 15*time.Minute)
}
// GetReadmeCacheTTL returns the README cache TTL (default: 1h)
func GetReadmeCacheTTL() time.Duration {
return GetDurationOrDefault("ATCR_README_CACHE_TTL", 1*time.Hour)
}
// GetHealthCheckInterval returns the hold health check refresh interval (default: 15m)
func GetHealthCheckInterval() time.Duration {
return GetDurationOrDefault("ATCR_HEALTH_CHECK_INTERVAL", 15*time.Minute)
}
// Jetstream Configuration
// GetJetstreamURL returns the Jetstream WebSocket URL (default: wss://jetstream2.us-west.bsky.network/subscribe)
func GetJetstreamURL() string {
return GetEnvOrDefault("JETSTREAM_URL", "wss://jetstream2.us-west.bsky.network/subscribe")
}
// GetBackfillEnabled returns whether backfill is enabled (default: true)
func GetBackfillEnabled() bool {
// Backfill is enabled unless explicitly set to "false"
return os.Getenv("ATCR_BACKFILL_ENABLED") != "false"
}
// GetRelayEndpoint returns the relay endpoint for sync API (default: https://relay1.us-east.bsky.network)
func GetRelayEndpoint() string {
return GetEnvOrDefault("ATCR_RELAY_ENDPOINT", "https://relay1.us-east.bsky.network")
}
// GetBackfillInterval returns the backfill interval (default: 1h)
func GetBackfillInterval() time.Duration {
return GetDurationOrDefault("ATCR_BACKFILL_INTERVAL", 1*time.Hour)
}
// Test Mode Configuration
// GetTestMode returns whether test mode is enabled (default: false)
// Test mode enables HTTP for local DID resolution and transition:generic scope
func GetTestMode() bool {
return os.Getenv("TEST_MODE") == "true"
}
pkg/appview/config_test.go
+32 -1339
View File
File diff suppressed because it is too large Load Diff
pkg/hold/oci/http_helpers.go
+2 -1
View File
@@ -6,6 +6,7 @@ package oci
import (
"encoding/json"
"fmt"
"log/slog"
"net/http"
)
@@ -25,7 +26,7 @@ func RespondJSON(w http.ResponseWriter, status int, v any) {
if err := json.NewEncoder(w).Encode(v); err != nil {
// If encoding fails, we can't do much since headers are already sent
// Log the error but don't try to send another response
fmt.Printf("ERROR: failed to encode JSON response: %v\n", err)
slog.Error("Failed to encode JSON response", "error", err)
}
}
pkg/hold/oci/multipart.go
+59 -21
View File
@@ -5,7 +5,7 @@ import (
"crypto/sha256"
"encoding/hex"
"fmt"
"log"
"log/slog"
"sort"
"strings"
"sync"
@@ -97,7 +97,9 @@ func (m *MultipartManager) cleanupExpiredSessions() {
now := time.Now()
for uploadID, session := range m.sessions {
if now.Sub(session.LastActivity) > 24*time.Hour {
log.Printf("Cleaning up expired multipart session: uploadID=%s, age=%v", uploadID, now.Sub(session.CreatedAt))
slog.Debug("Cleaning up expired multipart session",
"uploadID", uploadID,
"age", now.Sub(session.CreatedAt))
delete(m.sessions, uploadID)
}
}
@@ -121,7 +123,10 @@ func (m *MultipartManager) CreateSession(digest string, mode MultipartMode, s3Up
m.sessions[uploadID] = session
m.mu.Unlock()
log.Printf("Created multipart session: uploadID=%s, digest=%s, mode=%v", uploadID, digest, mode)
slog.Debug("Created multipart session",
"uploadID", uploadID,
"digest", digest,
"mode", mode)
return session
}
@@ -144,7 +149,7 @@ func (m *MultipartManager) DeleteSession(uploadID string) {
defer m.mu.Unlock()
delete(m.sessions, uploadID)
log.Printf("Deleted multipart session: uploadID=%s", uploadID)
slog.Debug("Deleted multipart session", "uploadID", uploadID)
}
// StorePart stores a part in the session (for Buffered mode)
@@ -167,7 +172,11 @@ func (s *MultipartSession) StorePart(partNumber int, data []byte) string {
s.Parts[partNumber] = part
s.LastActivity = time.Now()
log.Printf("Stored part: uploadID=%s, part=%d, size=%d bytes, etag=%s", s.UploadID, partNumber, len(data), etag)
slog.Debug("Stored part",
"uploadID", s.UploadID,
"part", partNumber,
"size", len(data),
"etag", etag)
return etag
}
@@ -205,7 +214,10 @@ func (s *MultipartSession) AssembleBufferedParts() ([]byte, int64, error) {
assembled = append(assembled, part.Data...)
}
log.Printf("Assembled buffered parts: uploadID=%s, parts=%d, totalSize=%d bytes", s.UploadID, maxPart, totalSize)
slog.Debug("Assembled buffered parts",
"uploadID", s.UploadID,
"parts", maxPart,
"totalSize", totalSize)
return assembled, totalSize, nil
}
@@ -214,9 +226,9 @@ func (s *MultipartSession) AssembleBufferedParts() ([]byte, int64, error) {
func (h *XRPCHandler) StartMultipartUploadWithManager(ctx context.Context, digest string) (string, MultipartMode, error) {
// Check if presigned URLs are disabled for testing
if h.disablePresignedURLs {
log.Printf("Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode")
slog.Debug("Presigned URLs disabled, using buffered mode", "reason", "DISABLE_PRESIGNED_URLS=true")
session := h.MultipartMgr.CreateSession(digest, Buffered, "")
log.Printf("Started buffered multipart: uploadID=%s", session.UploadID)
slog.Debug("Started buffered multipart", "uploadID", session.UploadID)
return session.UploadID, Buffered, nil
}
@@ -239,15 +251,18 @@ func (h *XRPCHandler) StartMultipartUploadWithManager(ctx context.Context, diges
s3UploadID := *result.UploadId
// S3 native multipart succeeded
session := h.MultipartMgr.CreateSession(digest, S3Native, s3UploadID)
log.Printf("Started S3 native multipart: digest=%s, uploadID=%s, s3UploadID=%s", digest, session.UploadID, s3UploadID)
slog.Debug("Started S3 native multipart",
"digest", digest,
"uploadID", session.UploadID,
"s3UploadID", s3UploadID)
return session.UploadID, S3Native, nil
}
log.Printf("S3 native multipart failed, falling back to buffered mode: %v", err)
slog.Warn("S3 native multipart failed, falling back to buffered mode", "error", err)
}
// Fallback to buffered mode
session := h.MultipartMgr.CreateSession(digest, Buffered, "")
log.Printf("Started buffered multipart: uploadID=%s", session.UploadID)
slog.Debug("Started buffered multipart", "uploadID", session.UploadID)
return session.UploadID, Buffered, nil
}
@@ -283,7 +298,10 @@ func (h *XRPCHandler) GetPartUploadURL(ctx context.Context, uploadID string, par
return nil, err
}
log.Printf("Generated part presigned URL: digest=%s, uploadID=%s, part=%d", session.Digest, uploadID, partNumber)
slog.Debug("Generated part presigned URL",
"digest", session.Digest,
"uploadID", uploadID,
"part", partNumber)
return &PartUploadInfo{
URL: url,
@@ -350,26 +368,40 @@ func (h *XRPCHandler) CompleteMultipartUploadWithManager(ctx context.Context, up
if err != nil {
return fmt.Errorf("failed to complete multipart upload: digest=%s, uploadID=%s, err=%v", session.Digest, uploadID, err)
}
log.Printf("Completed S3 native multipart at temp location: digest=%s, uploadID=%s, parts=%d", session.Digest, session.UploadID, len(s3Parts))
slog.Info("Completed S3 native multipart at temp location",
"digest", session.Digest,
"uploadID", session.UploadID,
"parts", len(s3Parts))
// Verify the blob exists at temp location before moving
destPath := blobPath(finalDigest)
log.Printf("[DEBUG] About to move: source=%s, dest=%s", sourcePath, destPath)
slog.Debug("About to move blob",
"source", sourcePath,
"dest", destPath)
if _, err := h.driver.Stat(ctx, sourcePath); err != nil {
log.Printf("[ERROR] Source blob not found after multipart complete: path=%s, err=%v", sourcePath, err)
slog.Error("Source blob not found after multipart complete",
"path", sourcePath,
"error", err)
return fmt.Errorf("source blob not found after multipart complete: %w", err)
}
log.Printf("[DEBUG] Source blob verified at: %s", sourcePath)
slog.Debug("Source blob verified", "path", sourcePath)
// Move from temp to final digest location using driver
// Driver handles path management correctly (including S3 prefix)
if err := h.driver.Move(ctx, sourcePath, destPath); err != nil {
log.Printf("[ERROR] Failed to move blob: source=%s, dest=%s, err=%v", sourcePath, destPath, err)
slog.Error("Failed to move blob",
"source", sourcePath,
"dest", destPath,
"error", err)
return fmt.Errorf("failed to move blob to final location: %w", err)
}
log.Printf("Moved blob to final location: %s → %s (driver paths: %s → %s)", session.Digest, finalDigest, sourcePath, destPath)
slog.Info("Moved blob to final location",
"from", session.Digest,
"to", finalDigest,
"sourcePath", sourcePath,
"destPath", destPath)
return nil
}
@@ -396,7 +428,11 @@ func (h *XRPCHandler) CompleteMultipartUploadWithManager(ctx context.Context, up
return fmt.Errorf("failed to commit blob: %w", err)
}
log.Printf("Completed buffered multipart: uploadID=%s, finalDigest=%s, size=%d bytes, written=%d", session.UploadID, finalDigest, size, written)
slog.Info("Completed buffered multipart",
"uploadID", session.UploadID,
"finalDigest", finalDigest,
"size", size,
"written", written)
return nil
}
@@ -427,12 +463,14 @@ func (h *XRPCHandler) AbortMultipartUploadWithManager(ctx context.Context, uploa
if err != nil {
return fmt.Errorf("failed to abort multipart upload: digest=%s, uploadID=%s, err=%v", session.Digest, uploadID, err)
}
log.Printf("Aborted S3 native multipart: digest=%s, uploadID=%s", session.Digest, session.UploadID)
slog.Debug("Aborted S3 native multipart",
"digest", session.Digest,
"uploadID", session.UploadID)
return nil
}
// Buffered mode: just delete the session (parts are in memory)
log.Printf("Aborted buffered multipart: uploadID=%s", session.UploadID)
slog.Debug("Aborted buffered multipart", "uploadID", session.UploadID)
return nil
}
pkg/hold/oci/xrpc.go
+3 -2
View File
@@ -3,6 +3,7 @@ package oci
import (
"fmt"
"io"
"log/slog"
"net/http"
"strconv"
@@ -268,7 +269,7 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques
_, _, err := h.pds.CreateLayerRecord(ctx, record)
if err != nil {
fmt.Printf("Failed to create layer record: %v\n", err)
slog.Error("Failed to create layer record", "error", err)
// Continue creating other records
} else {
layersCreated++
@@ -302,7 +303,7 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques
totalSize,
)
if err != nil {
fmt.Printf("Failed to create manifest post: %v\n", err)
slog.Error("Failed to create manifest post", "error", err)
} else {
postCreated = true
}
pkg/hold/pds/auth.go
+3 -3
View File
@@ -6,7 +6,7 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"log/slog"
"net/http"
"slices"
"strings"
@@ -426,7 +426,7 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient
return nil, fmt.Errorf("missing token")
}
log.Printf("[ValidateServiceToken] Validating service token for hold %s", holdDID)
slog.Debug("Validating service token", "holdDID", holdDID)
// Manually parse JWT (bypass golang-jwt since it doesn't support ES256K algorithm used by ATProto)
// Split token: header.payload.signature
@@ -493,7 +493,7 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient
return nil, fmt.Errorf("signature verification failed: %w", err)
}
log.Printf("[ValidateServiceToken] Successfully validated service token for user %s", issuerDID)
slog.Debug("Successfully validated service token", "userDID", issuerDID)
// Return validated user
return &ValidatedUser{
pkg/hold/pds/captain.go
+4 -1
View File
@@ -3,6 +3,7 @@ package pds
import (
"context"
"fmt"
"log/slog"
"time"
"atcr.io/pkg/atproto"
@@ -32,7 +33,9 @@ func (p *HoldPDS) CreateCaptainRecord(ctx context.Context, ownerDID string, publ
return cid.Undef, fmt.Errorf("failed to create captain record: %w", err)
}
fmt.Printf("Created captain record at %s, cid: %s\n", recordPath, recordCID)
slog.Info("Created captain record",
"path", recordPath,
"cid", recordCID.String())
return recordCID, nil
}
pkg/hold/pds/events.go
+37 -27
View File
@@ -6,7 +6,7 @@ import (
"database/sql"
"encoding/json"
"fmt"
"log"
"log/slog"
"strings"
"sync"
"time"
@@ -79,8 +79,8 @@ func NewEventBroadcaster(holdDID string, maxHistory int, dbPath string) *EventBr
// Initialize database connection and schema
if dbPath != "" && dbPath != ":memory:" {
if err := broadcaster.initDatabase(); err != nil {
log.Printf("Warning: Failed to initialize event database: %v", err)
log.Printf("Events will not persist across restarts")
slog.Warn("Failed to initialize event database", "error", err)
slog.Warn("Events will not persist across restarts")
}
}
@@ -127,14 +127,14 @@ func (b *EventBroadcaster) initDatabase() error {
var lastSeq sql.NullInt64
err = db.QueryRow("SELECT MAX(seq) FROM firehose_events").Scan(&lastSeq)
if err != nil {
log.Printf("Warning: Failed to load last event sequence: %v", err)
slog.Warn("Failed to load last event sequence", "error", err)
} else if lastSeq.Valid {
b.eventSeq = lastSeq.Int64
log.Printf("Loaded event sequence from database: seq=%d", b.eventSeq)
slog.Info("Loaded event sequence from database", "seq", b.eventSeq)
} else {
// Database is empty but might have existing repo records
// This happens on first deployment after adding persistent events
log.Printf("No events in database - will bootstrap from repo if needed")
slog.Info("No events in database - will bootstrap from repo if needed")
}
return nil
@@ -185,7 +185,7 @@ func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error {
}
if count > 0 {
log.Printf("Database already has %d events, skipping bootstrap", count)
slog.Info("Database already has events, skipping bootstrap", "count", count)
return nil
}
@@ -200,7 +200,7 @@ func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error {
head, err := pds.carstore.GetUserRepoHead(ctx, pds.uid)
if err != nil || !head.Defined() {
// Empty repo, nothing to bootstrap
log.Printf("Empty repo, no events to bootstrap")
slog.Info("Empty repo, no events to bootstrap")
return nil
}
@@ -215,7 +215,9 @@ func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error {
return fmt.Errorf("failed to get repo rev: %w", err)
}
log.Printf("Bootstrapping firehose events from current repo state (head=%s, rev=%s)", head.String(), rev)
slog.Info("Bootstrapping firehose events from current repo state",
"head", head.String(),
"rev", rev)
var recordCount int64
@@ -224,13 +226,13 @@ func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error {
// Get record value
_, recBytes, err := repoHandle.GetRecordBytes(ctx, path)
if err != nil {
log.Printf("Warning: failed to get record bytes for %s: %v", path, err)
slog.Warn("Failed to get record bytes", "path", path, "error", err)
return nil // Skip this record but continue
}
recordValue, err := lexutil.CborDecodeValue(*recBytes)
if err != nil {
log.Printf("Warning: failed to decode record %s: %v", path, err)
slog.Warn("Failed to decode record", "path", path, "error", err)
return nil
}
@@ -265,13 +267,13 @@ func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error {
Version: 1,
}
if err := car.WriteHeader(carHeader, &carBuf); err != nil {
log.Printf("Warning: failed to write CAR header: %v", err)
slog.Warn("Failed to write CAR header", "error", err)
return nil
}
// Write the record block
if err := carutil.LdWrite(&carBuf, recordCID.Bytes(), *recBytes); err != nil {
log.Printf("Warning: failed to write record block: %v", err)
slog.Warn("Failed to write record block", "error", err)
return nil
}
@@ -311,7 +313,9 @@ func (b *EventBroadcaster) BootstrapFromRepo(pds *HoldPDS) error {
return fmt.Errorf("failed to walk repo: %w", err)
}
log.Printf("✅ Bootstrapped %d events from repo (seq now at %d)", recordCount, b.eventSeq)
slog.Info("Bootstrapped events from repo",
"recordCount", recordCount,
"seq", b.eventSeq)
return nil
}
@@ -349,7 +353,9 @@ func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64) *Subscr
} else if cursor > currentSeq {
// Relay has cursor ahead of us - server was restarted
// Database should have the events if we had them before
log.Printf("Relay cursor %d > currentSeq %d (server restarted), attempting database backfill", cursor, currentSeq)
slog.Info("Relay cursor ahead of current seq, attempting database backfill",
"cursor", cursor,
"currentSeq", currentSeq)
go b.backfillSubscriber(sub, cursor)
}
// else cursor == currentSeq: relay is caught up, just stream new events
@@ -387,7 +393,9 @@ func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) {
// Persist event to database
if b.db != nil {
if err := b.persistEvent(commitEvent); err != nil {
log.Printf("Warning: Failed to persist event seq=%d to database: %v", seq, err)
slog.Warn("Failed to persist event to database",
"seq", seq,
"error", err)
}
}
@@ -401,7 +409,7 @@ func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) {
// Sent successfully
default:
// Subscriber's buffer is full, skip (they'll get disconnected for being too slow)
log.Printf("Warning: subscriber buffer full, skipping event seq=%d", seq)
slog.Warn("Subscriber buffer full, skipping event", "seq", seq)
}
}
}
@@ -488,7 +496,7 @@ func (b *EventBroadcaster) backfillSubscriber(sub *Subscriber, cursor int64) {
// If database is available, use it for backfill
if b.db != nil {
if err := b.backfillFromDatabase(sub, cursor); err != nil {
log.Printf("Database backfill failed, falling back to in-memory: %v", err)
slog.Warn("Database backfill failed, falling back to in-memory", "error", err)
b.backfillFromMemory(sub, cursor)
}
return
@@ -527,14 +535,14 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e
)
if err := rows.Scan(&seq, &commitCID, &rev, &sinceRev, &repoSlice, &opsJSON, &createdAt); err != nil {
log.Printf("Error scanning event row: %v", err)
slog.Error("Error scanning event row", "error", err)
continue
}
// Deserialize ops from JSON
var ops []*atproto.SyncSubscribeRepos_RepoOp
if err := json.Unmarshal(opsJSON, &ops); err != nil {
log.Printf("Error unmarshaling ops for seq=%d: %v", seq, err)
slog.Error("Error unmarshaling ops", "seq", seq, "error", err)
continue
}
@@ -562,7 +570,7 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e
// Sent successfully
case <-time.After(5 * time.Second):
// Timeout, subscriber too slow
log.Printf("Backfill timeout for subscriber at seq=%d", seq)
slog.Warn("Backfill timeout for subscriber", "seq", seq)
return nil
}
}
@@ -582,7 +590,7 @@ func (b *EventBroadcaster) backfillFromMemory(sub *Subscriber, cursor int64) {
// Sent
case <-time.After(5 * time.Second):
// Timeout, subscriber too slow
log.Printf("Backfill timeout for subscriber at seq=%d", he.Seq)
slog.Warn("Backfill timeout for subscriber", "seq", he.Seq)
return
}
}
@@ -606,13 +614,13 @@ func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) {
// Get a writer for this message
wc, err := sub.conn.NextWriter(websocket.BinaryMessage)
if err != nil {
log.Printf("Failed to get websocket writer: %v", err)
slog.Error("Failed to get websocket writer", "error", err)
return
}
// Write header as CBOR
if err := header.MarshalCBOR(wc); err != nil {
log.Printf("Failed to write event header: %v", err)
slog.Error("Failed to write event header", "error", err)
wc.Close()
return
}
@@ -623,14 +631,14 @@ func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) {
// Write the event as CBOR
var obj lexutil.CBOR = indigoEvent
if err := obj.MarshalCBOR(wc); err != nil {
log.Printf("Failed to write event body: %v", err)
slog.Error("Failed to write event body", "error", err)
wc.Close()
return
}
// Close the writer to flush the message
if err := wc.Close(); err != nil {
log.Printf("Failed to close websocket writer: %v", err)
slog.Error("Failed to close websocket writer", "error", err)
return
}
@@ -645,7 +653,9 @@ func convertToIndigoCommit(event *RepoCommitEvent) *atproto.SyncSubscribeRepos_C
// Parse commit CID string to cid.Cid, then convert to LexLink
commitCID, err := cid.Decode(event.Commit)
if err != nil {
log.Printf("Warning: failed to parse commit CID %s: %v", event.Commit, err)
slog.Warn("Failed to parse commit CID",
"cid", event.Commit,
"error", err)
// Create an empty CID as fallback
commitCID = cid.Undef
}
pkg/hold/pds/keys.go
+4 -3
View File
@@ -2,6 +2,7 @@ package pds
import (
"fmt"
"log/slog"
"os"
"path/filepath"
@@ -42,7 +43,7 @@ func generateKey(keyPath string) (*atcrypto.PrivateKeyK256, error) {
return nil, fmt.Errorf("failed to write key file: %w", err)
}
fmt.Printf("Generated new K-256 signing key at %s\n", keyPath)
slog.Info("Generated new K-256 signing key", "path", keyPath)
return privateKey, nil
}
@@ -59,7 +60,7 @@ func loadKey(keyPath string) (*atcrypto.PrivateKeyK256, error) {
if err != nil {
// Check if this is an old P-256 PEM key (migration)
if isPEMFormat(keyBytes) {
fmt.Printf("⚠️ Detected old P-256 key, replacing with K-256...\n")
slog.Warn("Detected old P-256 key, replacing with K-256")
// Generate new K-256 key (overwrites old P-256)
return generateKey(keyPath)
}
@@ -67,7 +68,7 @@ func loadKey(keyPath string) (*atcrypto.PrivateKeyK256, error) {
return nil, fmt.Errorf("failed to parse private key: %w", err)
}
fmt.Printf("Loaded existing K-256 signing key from %s\n", keyPath)
slog.Info("Loaded existing K-256 signing key", "path", keyPath)
return privateKey, nil
}
pkg/hold/pds/manifest_post.go
+4 -1
View File
@@ -3,6 +3,7 @@ package pds
import (
"context"
"fmt"
"log/slog"
"strings"
"time"
@@ -55,7 +56,9 @@ func (p *HoldPDS) CreateManifestPost(
// Build ATProto URI for the post
postURI := fmt.Sprintf("at://%s/app.bsky.feed.post/%s", p.did, rkey)
fmt.Printf("Created manifest post: %s (cid: %s)\n", postURI, recordCID)
slog.Info("Created manifest post",
"uri", postURI,
"cid", recordCID.String())
return postURI, nil
}
pkg/hold/pds/profile.go
+12 -5
View File
@@ -6,6 +6,7 @@ import (
"crypto/sha256"
"fmt"
"io"
"log/slog"
"net/http"
"time"
@@ -137,20 +138,22 @@ func (p *HoldPDS) CreateProfileRecord(ctx context.Context, storageDriver driver.
// Download and upload avatar if URL is provided
if avatarURL != "" {
fmt.Printf("Downloading avatar from %s\n", avatarURL)
slog.Debug("Downloading avatar", "url", avatarURL)
imageData, mimeType, err := downloadImage(ctx, avatarURL)
if err != nil {
return cid.Undef, fmt.Errorf("failed to download avatar: %w", err)
}
fmt.Printf("Uploading avatar blob (%d bytes, %s)\n", len(imageData), mimeType)
slog.Debug("Uploading avatar blob",
"size", len(imageData),
"mimeType", mimeType)
avatarBlob, err := uploadBlobToStorage(ctx, storageDriver, p.did, imageData, mimeType)
if err != nil {
return cid.Undef, fmt.Errorf("failed to upload avatar blob: %w", err)
}
profile.Avatar = avatarBlob
fmt.Printf("Avatar uploaded successfully: %s\n", avatarBlob.Ref.String())
slog.Info("Avatar uploaded successfully", "ref", avatarBlob.Ref.String())
}
// Use repomgr.PutRecord - creates with explicit rkey, fails if already exists
@@ -159,7 +162,9 @@ func (p *HoldPDS) CreateProfileRecord(ctx context.Context, storageDriver driver.
return cid.Undef, fmt.Errorf("failed to create profile record: %w", err)
}
fmt.Printf("Created profile record at %s, cid: %s\n", recordPath, recordCID)
slog.Info("Created profile record",
"path", recordPath,
"cid", recordCID.String())
return recordCID, nil
}
@@ -200,7 +205,9 @@ func (p *HoldPDS) CreateTangledProfileRecord(ctx context.Context, links []string
return cid.Undef, fmt.Errorf("failed to create tangled profile record: %w", err)
}
fmt.Printf("Created tangled profile record at %s, cid: %s\n", recordPath, recordCID)
slog.Info("Created tangled profile record",
"path", recordPath,
"cid", recordCID.String())
return recordCID, nil
}
pkg/hold/pds/server.go
+16 -9
View File
@@ -3,6 +3,7 @@ package pds
import (
"context"
"fmt"
"log/slog"
"os"
"path/filepath"
"strings"
@@ -92,7 +93,7 @@ func NewHoldPDS(ctx context.Context, did, publicURL, dbPath, keyPath string, ena
// Initialize empty repo with first commit
// RepoManager requires at least one commit to exist
// We'll create this by doing a dummy operation in Bootstrap
fmt.Printf("New hold repo - will be initialized in Bootstrap\n")
slog.Info("New hold repo - will be initialized in Bootstrap")
}
return &HoldPDS{
@@ -134,9 +135,9 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri
if captainExists {
// Captain record exists, skip captain/crew setup but still create profile if needed
fmt.Printf("✅ Captain record exists, skipping captain/crew setup\n")
slog.Info("Captain record exists, skipping captain/crew setup")
} else {
fmt.Printf("🚀 Bootstrapping hold PDS with owner: %s\n", ownerDID)
slog.Info("Bootstrapping hold PDS", "owner", ownerDID)
}
if !captainExists {
@@ -151,7 +152,7 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri
if err != nil {
return fmt.Errorf("failed to initialize repo: %w", err)
}
fmt.Printf("✅ Initialized empty repo\n")
slog.Info("Initialized empty repo")
}
// Create captain record (hold ownership and settings)
@@ -160,7 +161,10 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri
return fmt.Errorf("failed to create captain record: %w", err)
}
fmt.Printf("✅ Created captain record (public=%v, allowAllCrew=%v, enableBlueskyPosts=%v)\n", public, allowAllCrew, p.enableBlueskyPosts)
slog.Info("Created captain record",
"public", public,
"allowAllCrew", allowAllCrew,
"enableBlueskyPosts", p.enableBlueskyPosts)
// Add hold owner as first crew member with admin role
_, err = p.AddCrewMember(ctx, ownerDID, "admin", []string{"blob:read", "blob:write", "crew:admin"})
@@ -168,7 +172,7 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri
return fmt.Errorf("failed to add owner as crew member: %w", err)
}
fmt.Printf("✅ Added %s as hold admin\n", ownerDID)
slog.Info("Added owner as hold admin", "did", ownerDID)
} else {
// Captain record exists, check if we need to sync settings from env vars
_, existingCaptain, err := p.GetCaptainRecord(ctx)
@@ -184,7 +188,10 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri
if err != nil {
return fmt.Errorf("failed to update captain record: %w", err)
}
fmt.Printf("✅ Synced captain record with env vars (public=%v, allowAllCrew=%v, enableBlueskyPosts=%v)\n", public, allowAllCrew, p.enableBlueskyPosts)
slog.Info("Synced captain record with env vars",
"public", public,
"allowAllCrew", allowAllCrew,
"enableBlueskyPosts", p.enableBlueskyPosts)
}
}
}
@@ -203,9 +210,9 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri
if err != nil {
return fmt.Errorf("failed to create bluesky profile record: %w", err)
}
fmt.Printf("✅ Created Bluesky profile record (displayName=%s)\n", displayName)
slog.Info("Created Bluesky profile record", "displayName", displayName)
} else {
fmt.Printf("✅ Bluesky profile record already exists, skipping\n")
slog.Info("Bluesky profile record already exists, skipping")
}
}
pkg/hold/pds/status.go
+7 -2
View File
@@ -3,6 +3,7 @@ package pds
import (
"context"
"fmt"
"log/slog"
"time"
bsky "github.com/bluesky-social/indigo/api/bsky"
@@ -19,7 +20,7 @@ const (
func (p *HoldPDS) SetStatus(ctx context.Context, status string) error {
// Check if Bluesky posts are enabled
if !p.enableBlueskyPosts {
fmt.Printf("Bluesky posts disabled, skipping status post: %s\n", status)
slog.Debug("Bluesky posts disabled, skipping status post", "status", status)
return nil
}
@@ -51,6 +52,10 @@ func (p *HoldPDS) createStatusPost(ctx context.Context, text string) error {
return fmt.Errorf("failed to create status post: %w", err)
}
fmt.Printf("Created status post at %s/%s (rkey: %s), cid: %s, text: %s\n", StatusPostCollection, rkey, rkey, recordCID, text)
slog.Info("Created status post",
"collection", StatusPostCollection,
"rkey", rkey,
"cid", recordCID.String(),
"text", text)
return nil
}
pkg/hold/pds/xrpc.go
+56 -27
View File
@@ -20,7 +20,7 @@ import (
"crypto/sha256"
"io"
"log"
"log/slog"
"net/http"
"strconv"
"strings"
@@ -827,7 +827,7 @@ func (h *XRPCHandler) HandleGetRepo(w http.ResponseWriter, r *http.Request) {
if err != nil {
// Error already written to response by ReadRepo streaming
// Log it but don't try to write another HTTP error
fmt.Printf("Error streaming repo CAR: %v\n", err)
slog.Error("Error streaming repo CAR", "error", err)
return
}
}
@@ -865,7 +865,7 @@ func (h *XRPCHandler) HandleSubscribeRepos(w http.ResponseWriter, r *http.Reques
// Upgrade to WebSocket
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
fmt.Printf("WebSocket upgrade failed: %v\n", err)
slog.Error("WebSocket upgrade failed", "error", err)
return
}
@@ -970,7 +970,10 @@ func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) {
did := r.URL.Query().Get("did")
cidOrDigest := r.URL.Query().Get("cid")
log.Printf("[HandleGetBlob] %s request - did=%s, cid=%s", r.Method, did, cidOrDigest)
slog.Debug("HandleGetBlob request",
"method", r.Method,
"did", did,
"cid", cidOrDigest)
if did == "" || cidOrDigest == "" {
http.Error(w, "missing required parameters", http.StatusBadRequest)
@@ -992,14 +995,14 @@ func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) {
// Returns JSON with presigned URL for AppView integration
// Authorization: Protected by hold access control (captain.public or crew with blob:read)
func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, did, digest string) {
log.Printf("[handleGetOCIBlob] Processing OCI blob: %s", digest)
slog.Debug("Processing OCI blob", "digest", digest)
// Validate blob read access (hold access control)
// If captain.public = true, returns nil (public access allowed)
// If captain.public = false, validates auth and checks for blob:read permission
_, err := ValidateBlobReadAccess(r, h.pds, h.httpClient)
if err != nil {
log.Printf("[handleGetOCIBlob] Authorization failed: %v", err)
slog.Warn("OCI blob authorization failed", "error", err, "digest", digest)
http.Error(w, fmt.Sprintf("authorization failed: %v", err), http.StatusForbidden)
return
}
@@ -1014,12 +1017,18 @@ func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, d
// Generate presigned URL (use empty DID for content-addressed storage)
presignedURL, err := h.GetPresignedURL(r.Context(), operation, digest, "")
if err != nil {
log.Printf("[handleGetOCIBlob] Failed to get presigned %s URL: %v", operation, err)
slog.Error("Failed to get presigned URL for OCI blob",
"error", err,
"operation", operation,
"digest", digest)
http.Error(w, "failed to get presigned URL", http.StatusInternalServerError)
return
}
log.Printf("[handleGetOCIBlob] Returning presigned %s URL: %s", operation, presignedURL)
slog.Debug("Returning presigned URL for OCI blob",
"operation", operation,
"digest", digest,
"url", presignedURL)
// Return JSON response with presigned URL (AppView expects this format)
response := map[string]string{
@@ -1033,11 +1042,13 @@ func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, d
// Returns 307 redirect to presigned URL (standard ATProto behavior)
// Authorization: Public per ATProto spec (no auth required)
func (h *XRPCHandler) handleGetATProtoBlob(w http.ResponseWriter, r *http.Request, did, cid string) {
log.Printf("[handleGetATProtoBlob] Processing ATProto blob: %s", cid)
slog.Debug("Processing ATProto blob", "cid", cid)
// Validate DID (ATProto blobs are stored per-DID for data sovereignty)
if did != h.pds.DID() {
log.Printf("[handleGetATProtoBlob] DID mismatch: got %s, expected %s", did, h.pds.DID())
slog.Warn("ATProto blob DID mismatch",
"got", did,
"expected", h.pds.DID())
http.Error(w, "invalid did", http.StatusBadRequest)
return
}
@@ -1051,7 +1062,11 @@ func (h *XRPCHandler) handleGetATProtoBlob(w http.ResponseWriter, r *http.Reques
// Generate presigned URL (use DID for per-DID storage path)
presignedURL, err := h.GetPresignedURL(r.Context(), operation, cid, did)
if err != nil {
log.Printf("[handleGetATProtoBlob] Failed to get presigned %s URL: %v", operation, err)
slog.Error("Failed to get presigned URL for ATProto blob",
"error", err,
"operation", operation,
"cid", cid,
"did", did)
http.Error(w, "failed to get presigned URL", http.StatusInternalServerError)
return
}
@@ -1170,7 +1185,7 @@ func (h *XRPCHandler) HandleAtprotoDID(w http.ResponseWriter, r *http.Request) {
// This endpoint allows authenticated users to request crew membership
// Authorization is checked against captain record settings
func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request) {
log.Printf("[HandleRequestCrew] Starting crew membership request")
slog.Debug("Starting crew membership request")
// Get authenticated user from context (if coming through middleware)
// Otherwise validate directly (for tests or direct handler calls)
@@ -1179,12 +1194,12 @@ func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request)
var err error
user, err = ValidateDPoPRequest(r, h.httpClient)
if err != nil {
log.Printf("[HandleRequestCrew] Authentication failed: %v", err)
slog.Warn("Crew request authentication failed", "error", err)
http.Error(w, fmt.Sprintf("authentication failed: %v", err), http.StatusUnauthorized)
return
}
}
log.Printf("[HandleRequestCrew] Authenticated user: %s", user.DID)
slog.Debug("Authenticated user for crew request", "did", user.DID)
// Parse request body (optional parameters)
var req struct {
@@ -1195,21 +1210,23 @@ func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request)
// Body is optional - if empty, just use defaults
if r.Body != nil && r.ContentLength > 0 {
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
log.Printf("[HandleRequestCrew] Failed to parse request body: %v", err)
slog.Warn("Failed to parse crew request body", "error", err)
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
}
// Get captain record to check authorization settings
log.Printf("[HandleRequestCrew] Getting captain record...")
slog.Debug("Getting captain record for crew request")
_, captain, err := h.pds.GetCaptainRecord(r.Context())
if err != nil {
log.Printf("[HandleRequestCrew] Failed to get captain record: %v", err)
slog.Error("Failed to get captain record", "error", err)
http.Error(w, fmt.Sprintf("failed to get captain record: %v", err), http.StatusInternalServerError)
return
}
log.Printf("[HandleRequestCrew] Captain record retrieved: owner=%s, allowAllCrew=%v", captain.Owner, captain.AllowAllCrew)
slog.Debug("Captain record retrieved",
"owner", captain.Owner,
"allowAllCrew", captain.AllowAllCrew)
// Check authorization:
// 1. If allowAllCrew is true, any authenticated user can join
@@ -1231,19 +1248,21 @@ func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request)
// Check if user is already a crew member
// List all crew members and check if this DID is already present
log.Printf("[HandleRequestCrew] Checking existing crew membership...")
slog.Debug("Checking existing crew membership")
crew, err := h.pds.ListCrewMembers(r.Context())
if err != nil {
log.Printf("[HandleRequestCrew] Failed to list crew members: %v", err)
slog.Error("Failed to list crew members", "error", err)
http.Error(w, fmt.Sprintf("failed to list crew members: %v", err), http.StatusInternalServerError)
return
}
log.Printf("[HandleRequestCrew] Found %d existing crew members", len(crew))
slog.Debug("Found existing crew members", "count", len(crew))
for _, member := range crew {
if member.Record.Member == user.DID {
// Already a crew member, return success with existing record
log.Printf("[HandleRequestCrew] User is already a crew member (rkey=%s)", member.Rkey)
slog.Debug("User is already a crew member",
"did", user.DID,
"rkey", member.Rkey)
response := map[string]any{
"uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), atproto.CrewCollection, member.Rkey),
"cid": member.Cid.String(),
@@ -1258,14 +1277,21 @@ func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request)
}
// Create new crew record
log.Printf("[HandleRequestCrew] Creating new crew record for user %s (role=%s, permissions=%v)", user.DID, req.Role, req.Permissions)
slog.Debug("Creating new crew record",
"did", user.DID,
"role", req.Role,
"permissions", req.Permissions)
recordCID, err := h.pds.AddCrewMember(r.Context(), user.DID, req.Role, req.Permissions)
if err != nil {
log.Printf("[HandleRequestCrew] Failed to create crew record: %v", err)
slog.Error("Failed to create crew record",
"error", err,
"did", user.DID)
http.Error(w, fmt.Sprintf("failed to create crew record: %v", err), http.StatusInternalServerError)
return
}
log.Printf("[HandleRequestCrew] Successfully created crew record (CID=%s)", recordCID.String())
slog.Info("Successfully created crew record",
"did", user.DID,
"cid", recordCID.String())
// Return success response
// Note: rkey is generated by AddCrewMember (TID), we don't have direct access to it
@@ -1341,8 +1367,11 @@ func (h *XRPCHandler) GetPresignedURL(ctx context.Context, operation string, dig
// Generate presigned URL with 15 minute expiry
url, err := req.Presign(15 * time.Minute)
if err != nil {
log.Printf("[getPresignedURL] Presign FAILED for %s: %v", operation, err)
log.Printf(" Falling back to XRPC endpoint")
slog.Warn("Presign failed, falling back to XRPC endpoint",
"error", err,
"operation", operation,
"digest", digest)
slog.Debug("Using XRPC proxy fallback")
proxyURL := getProxyURL(h.pds.PublicURL, digest, did, operation)
if proxyURL == "" {
return "", fmt.Errorf("presign failed and XRPC proxy not supported for PUT operations")