Replaces the sequential worker with a download-manager-style pool (§0 item 4). Model: the user's connection is the limit; semantics are fair-share ACROSS feeds, not goroutine-per-book (a global FIFO would make Book B wait behind Book A's 100 files). - Atomic claim: GetNextQueuedEpisode + SetPodcastEpisodeDownloading was a read-then-write race under concurrency; ClaimQueuedPodcastEpisode converts queued→downloading in ONE guarded statement (two claimants, one winner — the loser gets ErrNoRows). SetPodcastEpisodeDownloading retired. - Queue tickets: download_queue_pos BIGINT fed by a Postgres SEQUENCE stamped at enqueue — the enqueue loop's ordering IS the claim order (the foundation §0 item 5 builds on; previously all rows in a batch shared one updated_at, so within-feed order was arbitrary). - Pool: up to podcast_download_concurrency (default 4, live range 1–16) claim-loop goroutines per drain round; the setting re-reads every round so admin changes take effect without restart. - Fair share: claim prefers the feed least-recently-served (pickFeedToServe; never-served first), earliest ticket within a feed — books proceed in parallel while chapters land in order (start listening while the rest download). - Bandwidth: shared token bucket across all workers (podcast_download_speed_mbps, default 0 = unlimited); take() reserves bandwidth and callers sleep outside the lock. - Settings: new float setting type end to end (registry defaults + clamping, save validation, admin UI number-with-step input) + the two schema seeds; worker wires the registry via SetSettings (tests run on registered defaults). - Tests: unit coverage for the bucket reservation math + fair-share pick; integration TestPodcastDownloadPool pins ticket order = enqueue order, claim exclusivity, and a full 3-episode pool drain. Dev-DB note: the column/sequence/settings were applied live (idempotent DDL, same statements the schema runs at next boot) since the running container predates them.
307 lines
12 KiB
Go
307 lines
12 KiB
Go
package main
|
|
|
|
import (
|
|
"bookhoard/internal/app"
|
|
"bookhoard/internal/config"
|
|
"bookhoard/internal/database"
|
|
"bookhoard/internal/handlers"
|
|
"bookhoard/internal/middleware"
|
|
ratelimit "bookhoard/internal/middleware"
|
|
"bookhoard/internal/router"
|
|
"bookhoard/internal/services"
|
|
"bookhoard/internal/sync"
|
|
"bookhoard/internal/version"
|
|
"bookhoard/templates"
|
|
"context"
|
|
"log"
|
|
"time"
|
|
|
|
_ "time/tzdata"
|
|
|
|
"github.com/go-playground/validator/v10"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"github.com/labstack/echo/v5"
|
|
echomiddleware "github.com/labstack/echo/v5/middleware"
|
|
)
|
|
|
|
// CustomValidator wraps the go-playground validator
|
|
type CustomValidator struct {
|
|
validator *validator.Validate
|
|
}
|
|
|
|
func (cv *CustomValidator) Validate(i interface{}) error {
|
|
return cv.validator.Struct(i)
|
|
}
|
|
|
|
func main() {
|
|
log.Printf("Bookhoard %s (commit %s) starting", version.Version, version.Commit)
|
|
cfg := config.LoadConfig()
|
|
|
|
dbPool, err := pgxpool.New(context.Background(), cfg.DatabaseURL())
|
|
if err != nil {
|
|
log.Fatal("Failed to connect to database:", err)
|
|
}
|
|
defer dbPool.Close()
|
|
|
|
// The search service's raw DBTX path must ALSO carry the RLS
|
|
// identity — hand it the ScopedDB itself (a DBTX), never the bare
|
|
// pool, or search queries bypass the per-user GUC and raise under
|
|
// the policies.
|
|
scopedDB := database.NewScopedDB(dbPool)
|
|
queries := database.New(scopedDB)
|
|
|
|
// Initialize database schema
|
|
log.Println("🔧 Ensuring database schema is initialized...")
|
|
ctx := context.Background()
|
|
if err := database.Initialize(ctx, dbPool); err != nil {
|
|
log.Fatal("❌ Database schema initialization failed:", err)
|
|
}
|
|
log.Println("✅ Database schema initialized and verified, starting server...")
|
|
|
|
// Load tunable settings from the DB into the registry. All values fall back
|
|
// to compiled defaults if a row is missing, so this never blocks startup.
|
|
// (Startup has no end-user identity; system_settings is not RLS-
|
|
// governed, and ScopedDB declares nothing when no identity exists —
|
|
// the engine only raises on governed tables.)
|
|
registry := database.NewSettingsRegistry(queries)
|
|
if err := registry.Load(ctx); err != nil {
|
|
log.Printf("⚠️ Could not load system settings (using defaults): %v", err)
|
|
}
|
|
|
|
// Public signup surfaces follow the registration_enabled flag, with the
|
|
// first-user exception: an unconfigured server (zero admins) keeps its
|
|
// signup links and registration route reachable for bootstrapping.
|
|
templates.SetSignupVisibility(func() bool {
|
|
if registry.RegistrationEnabled() {
|
|
return true
|
|
}
|
|
count, err := queries.CountAdmins(context.Background())
|
|
if err != nil {
|
|
return true // fail open, matching setupstatus
|
|
}
|
|
return count == 0
|
|
})
|
|
// Wire the registry into the package-level password validator so live
|
|
// rule changes apply to the echo struct-tag validator and ValidatePassword.
|
|
middleware.SetDefaultPasswordSettings(registry)
|
|
|
|
// Seed base_url from env var if not already configured. Uses conditional
|
|
// UPDATE so admin-set values are never overwritten on restart.
|
|
if cfg.BaseURL != "" {
|
|
_, err = dbPool.Exec(ctx, `
|
|
INSERT INTO system_config (key, value)
|
|
VALUES ('base_url', $1)
|
|
ON CONFLICT (key) DO UPDATE
|
|
SET value = EXCLUDED.value
|
|
WHERE system_config.value = ''
|
|
`, cfg.BaseURL)
|
|
if err != nil {
|
|
log.Printf("⚠️ Could not seed base_url: %v", err)
|
|
} else {
|
|
// Also seed derived URLs
|
|
for key, suffix := range map[string]string{
|
|
"opds_base_url": "/opds",
|
|
"api_base_url": "/api",
|
|
} {
|
|
_, _ = dbPool.Exec(ctx, `
|
|
INSERT INTO system_config (key, value)
|
|
VALUES ($1, $2)
|
|
ON CONFLICT (key) DO UPDATE
|
|
SET value = EXCLUDED.value
|
|
WHERE system_config.value = ''
|
|
`, key, cfg.BaseURL+suffix)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Create login attempt tracker from configured (or default) lockout policy.
|
|
loginMaxAttempts, loginLockout := registry.LoginLockout()
|
|
loginAttemptTracker := ratelimit.NewLoginAttemptTracker(loginMaxAttempts, loginLockout, 5*time.Minute)
|
|
|
|
authHandler := handlers.NewAuthHandler(queries, cfg.JWTSecret, loginAttemptTracker)
|
|
authHandler.SetSettings(registry)
|
|
systemSettingsHandler := handlers.NewSystemSettingsHandler(queries)
|
|
systemSettingsHandler.SetSettings(registry)
|
|
sidecarHandler := handlers.NewSidecarHandler(queries, cfg)
|
|
sidecarHandler.SetSettings(registry)
|
|
libraryHandler := handlers.NewLibraryHandler(queries)
|
|
deviceHandler := handlers.NewDeviceHandler(queries, cfg.JWTSecret, cfg)
|
|
deviceAuthMiddleware := middleware.NewDeviceAuthMiddleware(queries)
|
|
deviceAuthMiddleware.SetSettings(registry)
|
|
processingIssuesHandler := handlers.NewProcessingIssuesHandler(queries)
|
|
hashConflictsHandler := handlers.NewHashConflictsHandler(queries)
|
|
|
|
// Create WebSocket connection manager
|
|
connManager := sync.NewConnectionManager()
|
|
|
|
progressService := sync.NewProgressService(queries, connManager)
|
|
annotationService := sync.NewAnnotationService(queries, connManager)
|
|
annotationService.SetSettings(registry)
|
|
maintenanceCancel := annotationService.StartDailyMaintenance()
|
|
defer maintenanceCancel()
|
|
|
|
queueProcessor := sync.NewSyncQueueProcessorWithConfig(queries, progressService, annotationService, registry.SyncQueueConfig().Interval, registry.SyncQueueConfig().BatchSize)
|
|
|
|
// Create library service
|
|
libraryService := services.NewLibraryService(queries)
|
|
// The progress service verifies submitted anchors against the book
|
|
// itself — it needs to resolve library-relative file paths.
|
|
progressService.SetMediaPathResolver(libraryService)
|
|
|
|
// Sync Go AllowedExtensions into DB so API clients see correct extensions
|
|
libraryService.SyncAllowedExtensions(context.Background())
|
|
|
|
// Create worker for background tasks
|
|
workerCfg := registry.WorkerPoolConfig()
|
|
worker := services.NewWorkerWithConfig(workerCfg.Size, workerCfg.QueueCap, connManager)
|
|
services.WorkerInstance = worker
|
|
|
|
koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor, progressService, annotationService)
|
|
koreaderHandler.SetLibraryService(libraryService)
|
|
wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware)
|
|
conflictHandler := handlers.NewConflictHandler(queries, connManager, progressService)
|
|
analyticsHandler := handlers.NewAnalyticsHandler(queries)
|
|
statisticsService := services.NewStatisticsService(queries)
|
|
statsHandler := handlers.NewStatsHandler(statisticsService)
|
|
queueHandler := handlers.NewQueueHandler(queries, queueProcessor)
|
|
|
|
conversionService := services.NewConversionService(queries, "/var/bookhoard/cache/kepub")
|
|
conversionService.SetSettings(registry)
|
|
opdsHandler := handlers.NewOPDSHandler(queries, libraryService, conversionService)
|
|
opdsHandler.SetSettings(registry)
|
|
|
|
collectionHandler := handlers.NewCollectionHandler(queries, libraryService, connManager)
|
|
dashboardService := services.NewDashboardService(queries)
|
|
dashboardHandler := handlers.NewDashboardHandler(queries)
|
|
seriesHandler := handlers.NewSeriesHandler(queries)
|
|
filtersHandler := handlers.NewFiltersHandler(queries)
|
|
mediaHandler := handlers.NewMediaHandler(queries, scopedDB, libraryService, progressService, annotationService)
|
|
matchingHandler := handlers.NewMatchingHandler(queries, connManager)
|
|
jobsHandler := handlers.NewJobsHandler(queries, worker)
|
|
|
|
// Podcast subsystem: fetcher service + concurrent download worker
|
|
// (§0 item 4: pool, fair share, admin tunables) + poller. One
|
|
// instance trio per server (per-feed locks dedupe).
|
|
podcastService := services.NewPodcastService(queries)
|
|
podcastWorker := services.NewPodcastDownloadWorker(queries, podcastService)
|
|
podcastWorker.SetSettings(registry)
|
|
podcastHandler := handlers.NewPodcastHandler(queries, podcastService, podcastWorker)
|
|
|
|
e := echo.New()
|
|
|
|
// Set up validator
|
|
v := validator.New()
|
|
|
|
// Register custom password complexity validator
|
|
if err := ratelimit.RegisterPasswordValidation(v); err != nil {
|
|
log.Fatal("Failed to register password validator:", err)
|
|
}
|
|
|
|
e.Validator = &CustomValidator{validator: v}
|
|
|
|
// Middleware
|
|
e.Use(echomiddleware.RequestLogger())
|
|
e.Use(echomiddleware.Recover())
|
|
e.Use(echomiddleware.CORSWithConfig(echomiddleware.CORSConfig{
|
|
AllowOrigins: []string{"*"},
|
|
AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
|
|
AllowHeaders: []string{"Origin", "Content-Type", "Authorization"},
|
|
ExposeHeaders: []string{"Content-Length"},
|
|
AllowCredentials: false,
|
|
}))
|
|
e.Use(ratelimit.SecurityHeadersMiddleware())
|
|
e.Use(ratelimit.RequestTracingMiddleware(cfg))
|
|
|
|
// Rate limiter for auth endpoints
|
|
// rateLimiterConfig := ratelimit.RateLimiterConfig{
|
|
// Enabled: cfg.RateLimitEnabled,
|
|
// RequestsPerMinute: cfg.RequestsPerMinute,
|
|
// CleanupInterval: 5 * time.Minute,
|
|
// }
|
|
// rateLimiter := ratelimit.NewRateLimiter(rateLimiterConfig)
|
|
// rateLimitMiddleware := ratelimit.RateLimiterMiddleware(rateLimiter) // Now in router/auth.go
|
|
|
|
// ========================================================================
|
|
// ROUTER REGISTRATION - Migrate routes to internal/router/ package
|
|
// ========================================================================
|
|
routerConfig := &router.Config{
|
|
Echo: e,
|
|
Queries: queries,
|
|
Cfg: cfg,
|
|
Settings: registry,
|
|
DBPool: dbPool,
|
|
AuthHandler: authHandler,
|
|
LibraryHandler: libraryHandler,
|
|
DeviceHandler: deviceHandler,
|
|
MediaHandler: mediaHandler,
|
|
MatchingHandler: matchingHandler,
|
|
ProcessingIssuesHandler: processingIssuesHandler,
|
|
HashConflictsHandler: hashConflictsHandler,
|
|
KOReaderHandler: koreaderHandler,
|
|
WSHandler: wsHandler,
|
|
ConflictHandler: conflictHandler,
|
|
AnalyticsHandler: analyticsHandler,
|
|
StatsHandler: statsHandler,
|
|
StatisticsService: statisticsService,
|
|
QueueHandler: queueHandler,
|
|
CollectionHandler: collectionHandler,
|
|
FiltersHandler: filtersHandler,
|
|
DashboardHandler: dashboardHandler,
|
|
PodcastHandler: podcastHandler,
|
|
PodcastService: podcastService,
|
|
PodcastWorker: podcastWorker,
|
|
DashboardService: dashboardService,
|
|
SeriesHandler: seriesHandler,
|
|
OPDSHandler: opdsHandler,
|
|
Worker: worker,
|
|
SystemSettingsHandler: systemSettingsHandler,
|
|
SidecarHandler: sidecarHandler,
|
|
ConnManager: connManager,
|
|
QueueProcessor: queueProcessor,
|
|
ProgressService: progressService,
|
|
AnnotationService: annotationService,
|
|
DeviceAuthMiddleware: deviceAuthMiddleware,
|
|
JobsHandler: jobsHandler,
|
|
LoginTracker: loginAttemptTracker,
|
|
LibraryService: libraryService,
|
|
}
|
|
|
|
// Register all routes and get ebook handler
|
|
_ = router.RegisterRoutes(routerConfig)
|
|
|
|
// ========================================================================
|
|
// APPLICATION LIFECYCLE MANAGEMENT
|
|
// ========================================================================
|
|
|
|
// Create app with lifecycle management
|
|
application := app.New(e)
|
|
|
|
// Graceful shutdown: after the HTTP server drains, stop background work —
|
|
// drain job workers, release the device rate limiter's ticker, close scan
|
|
// log files — in registration order.
|
|
application.OnShutdown(func() { worker.Shutdown() })
|
|
application.OnShutdown(func() { deviceAuthMiddleware.Stop() })
|
|
application.OnShutdown(func() { routerConfig.ScannerHandler.CloseScanner() })
|
|
|
|
// Scan settings saves reach the scanner immediately instead of after its
|
|
// 30s settings-cache TTL.
|
|
systemSettingsHandler.SetScanSettingsChanged(func() {
|
|
routerConfig.ScannerHandler.InvalidateScannerSettingsCache()
|
|
})
|
|
|
|
// ========================================================================
|
|
// START SERVER (managed by app lifecycle)
|
|
// ========================================================================
|
|
log.Printf("Starting server on port %s", cfg.ServerPort)
|
|
|
|
// Start HTTP server
|
|
if err := application.StartServer(":" + cfg.ServerPort); err != nil {
|
|
log.Fatalf("Failed to start server: %v", err)
|
|
}
|
|
|
|
// Start application lifecycle (blocks until shutdown signal)
|
|
if err := application.Start(); err != nil {
|
|
log.Fatalf("Application error: %v", err)
|
|
}
|
|
}
|