Files
bookhoard/internal/handlers/scanner.go
T
John O'Keefe 5245f292ca feat(scanner): quiet scans, enqueue coalescing, watcher probe, sibling cache
Part 2 of the faceted plan, Phase 5 (steps 5-6): the scan that used to
print ~4 lines per unchanged file now prints ONE summary line, and a
poll tick landing mid-scan no longer stacks a second full scan.

- Quiet scans (§4): the walk loop is silent — inserts, changes, and
  errors log inside processMediaFile; the duplicate
  'Successfully processed' branch (both branches printed it) and the
  per-file 'processing / already exists / same size, skipping' chatter
  are gone. Summary carries the trigger and a changed counter:
  'Scan of <folders> (trigger: poll): N files, N changed, N new, Xs,
  N errors'. Triggers threaded through every enqueue site: poll
  (startBackupScan), watcher (dirty-dirs flush), admin (scan API),
  initial (startup scan).
- Enqueue coalescing (§5): Worker.HasActiveScanForLibrary (JobResult
  gains additive job_type so only SCAN jobs coalesce — an import
  sharing the library id never blocks one). enqueueLibraryScan skips
  silently; the admin scan API answers 409 with the active job id.
  Enforces the one-scan-per-library invariant the bulk scan-state map
  leans on.
- Watcher capability probe (§4): ONE Add on a root folder; failure
  disables the watcher with a single startup line (replaces the
  per-directory warning spam on inotify-broken deployments). The
  watcher stays an optional latency reducer over scheduled scans.
- Per-directory sibling cache (§6): hasSiblingBookFileCached caches
  the ReadDir per directory for one scan — loose-image manga re-stats
  thousands of images sharing a handful of directories.

Tests: TestHasActiveScanForLibrary (unit — running/pending coalesce,
terminal jobs don't, non-scan jobs don't, empty library never, nil
worker safe). Full scanner suite green in the container.
2026-10-19 20:35:07 -04:00

409 lines
13 KiB
Go

package handlers
import (
"bookhoard/internal/services"
"context"
"errors"
"fmt"
"net/http"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"github.com/labstack/echo/v5"
)
const (
maxPaginationLimit = 1000
)
// StartBackgroundTasks starts the queue processor and cleanup task
// This should be called once for the main handler instance
func (h *Handler) StartBackgroundTasks() {
h.cleanupTaskCancel = h.connManager.StartCleanupTask()
go h.queueProcessor.Start(h.queueCtx)
}
// ScanLibraryRequest represents the request for scanning a library
type ScanLibraryRequest struct {
FolderPaths []string `json:"folder_paths,omitempty"`
LibraryID string `json:"library_id,omitempty"`
Force bool `json:"force,omitempty"`
}
// ScanLibrary handles POST /api/scanner/scan (now runs in background)
func (h *Handler) ScanLibrary(c *echo.Context) error {
var req ScanLibraryRequest
// Check if scan_request is set in context (from library scan route)
if scanReq, ok := c.Get("scan_request").(map[string]interface{}); ok {
if libraryID, ok := scanReq["library_id"].(string); ok {
req.LibraryID = libraryID
}
}
// Bind request body if provided (for direct scanner/scan calls)
if err := c.Bind(&req); err != nil && req.LibraryID == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
// Get user ID from JWT token
userID := c.Get("user_id").(string)
userUUID, err := uuid.Parse(userID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid user id"})
}
var folderPaths []string
// If library_id is provided, fetch folders from library
if req.LibraryID != "" {
libraryUUID, err := uuid.Parse(req.LibraryID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid library id"})
}
// Fetch library folders from database
libraryFolders, err := h.db.GetLibraryFolders(c.Request().Context(), pgtype.UUID{Bytes: libraryUUID, Valid: true})
if err != nil {
return c.JSON(http.StatusNotFound, map[string]string{"error": "library not found or has no folders"})
}
// Extract folder paths
for _, folder := range libraryFolders {
folderPaths = append(folderPaths, folder.FolderPath)
}
if len(folderPaths) == 0 {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "library has no folders configured"})
}
} else if len(req.FolderPaths) > 0 {
// Use folder paths from request
folderPaths = req.FolderPaths
} else {
// Neither library_id nor folder_paths provided
return c.JSON(http.StatusBadRequest, map[string]string{"error": "either library_id or folder_paths required for scanning"})
}
// Coalescing (plan Part 2 §5): a scan for this library is already
// pending or running — 409 with the active job so the UI can show
// progress instead of stacking a second full scan.
if req.LibraryID != "" && h.worker.HasActiveScanForLibrary(req.LibraryID) {
var activeJobID string
for _, active := range h.worker.GetActiveScanJobs() {
if active.JobType == services.JobTypeScan && active.LibraryID == req.LibraryID {
activeJobID = active.JobID
break
}
}
return c.JSON(http.StatusConflict, map[string]string{
"error": "a scan for this library is already pending or running",
"active_job_id": activeJobID,
})
}
jobID := uuid.New().String()
job := &services.Job{
ID: jobID,
Type: services.JobTypeScan,
UserID: userID,
Params: map[string]interface{}{
"library_id": req.LibraryID,
"folders": folderPaths,
"admin_id": userUUID.String(),
"db": h.db,
"force": req.Force,
"trigger": "admin",
},
Status: services.JobStatusPending,
Context: h.ctx,
}
if err := h.worker.EnqueueJob(job); err != nil {
return c.JSON(http.StatusInternalServerError, map[string]string{"error": "failed to enqueue scan job: " + err.Error()})
}
return c.JSON(http.StatusAccepted, map[string]interface{}{
"message": "scan job enqueued",
"job_id": jobID,
"status": "pending",
})
}
// StartScanner handles POST /api/scanner/start
func (h *Handler) StartScanner(c *echo.Context) error {
h.mu.Lock()
defer h.mu.Unlock()
var req ScanLibraryRequest
if err := c.Bind(&req); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
// Get admin ID from JWT token
userID := c.Get("user_id").(string)
userUUID, err := uuid.Parse(userID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid user id"})
}
// Set the folder paths (watch=true: this long-lived scanner reads events)
if err := h.scanner.SetFolders(req.FolderPaths, true); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid folder paths: " + err.Error()})
}
// Set the admin ID for media item association
h.scanner.SetAdminID(pgtype.UUID{Bytes: userUUID, Valid: true})
// Start watching for changes
if !h.scanner.GetAutoScanEnabled() {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "auto-scan disabled in settings"})
}
h.scanner.WatchChanges(h.watchModeCtx)
return c.JSON(http.StatusOK, map[string]string{"message": "scanner started"})
}
// StopScanner handles POST /api/scanner/stop
func (h *Handler) StopScanner(c *echo.Context) error {
h.mu.Lock()
defer h.mu.Unlock()
h.watchModeCancel()
h.watchModeCtx, h.watchModeCancel = context.WithCancel(context.Background())
return c.JSON(http.StatusOK, map[string]string{"message": "scanner stopped"})
}
// GetScanStatus handles GET /api/scanner/status/:jobId
func (h *Handler) GetScanStatus(c *echo.Context) error {
jobID := c.Param("jobId")
result, exists := h.worker.GetJobStatus(jobID)
if !exists {
return c.JSON(http.StatusNotFound, map[string]string{"error": "job not found"})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"job_id": result.JobID,
"status": result.Status,
"error": result.Error,
"result": result.Result,
"progress": result.Progress,
"files_scanned": result.FilesScanned,
"new_items": result.NewItems,
"errors": result.Errors,
})
}
// GetActiveScans handles GET /api/scanner/active (admin): every
// pending/running job. Lets the admin page re-attach to in-flight
// scans after a page refresh instead of only tracking scans it
// started itself in this page load.
func (h *Handler) GetActiveScans(c *echo.Context) error {
jobs := h.worker.GetActiveScanJobs()
return c.JSON(http.StatusOK, map[string]interface{}{
"scanning": len(jobs) > 0,
"jobs": jobs,
})
}
// GetScanSummary handles GET /api/scanner/active-summary (any
// authenticated user): whether a scan is running and its average
// progress (0-1). Powers the sitewide sidebar indicator, which must
// survive page refreshes — websocket events alone only reach pages
// that were already open when the scan started.
func (h *Handler) GetScanSummary(c *echo.Context) error {
jobs := h.worker.GetActiveScanJobs()
progress := 0.0
if len(jobs) > 0 {
for _, job := range jobs {
progress += job.Progress
}
progress /= float64(len(jobs))
}
return c.JSON(http.StatusOK, map[string]interface{}{
"scanning": len(jobs) > 0,
"progress": progress,
})
}
// CancelScan handles POST /api/scanner/cancel/:jobId (admin): requests
// cancellation of a pending or running scan job. Cancellation is
// cooperative — the file currently being processed finishes, the walk
// stops at its next context check, and the archive pass is skipped so
// unvisited files are never treated as missing.
func (h *Handler) CancelScan(c *echo.Context) error {
jobID := c.Param("jobId")
if err := h.worker.CancelJob(jobID); err != nil {
if errors.Is(err, services.ErrJobNotFound) {
return c.JSON(http.StatusNotFound, map[string]string{"error": "job not found"})
}
return c.JSON(http.StatusConflict, map[string]string{"error": err.Error()})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"message": "scan cancellation requested",
"job_id": jobID,
"status": "cancelled",
})
}
// StartWatchModeForLibrary starts watching a specific library's folders
func (h *Handler) StartWatchModeForLibrary(ctx context.Context, libraryID pgtype.UUID, adminID pgtype.UUID) error {
h.mu.Lock()
defer h.mu.Unlock()
libraryIDStr := fmt.Sprintf("%x", libraryID.Bytes)
if h.watchingLibraries[libraryIDStr] {
return fmt.Errorf("already watching library %s", libraryIDStr)
}
folders, err := h.db.GetLibraryFolders(ctx, libraryID)
if err != nil {
return fmt.Errorf("failed to get library folders: %v", err)
}
if len(folders) == 0 {
return fmt.Errorf("no folders configured for library")
}
folderPaths := make([]string, len(folders))
for i, folder := range folders {
folderPaths[i] = folder.FolderPath
}
scanner := services.NewMediaScanner(h.db)
if err := scanner.SetFolders(folderPaths, true); err != nil {
return fmt.Errorf("failed to set scanner folders: %v", err)
}
scanner.SetAdminID(adminID)
scanner.SetLibraryID(libraryID)
scanner.WatchChanges(h.watchModeCtx)
h.watchingLibraries[libraryIDStr] = true
return nil
}
// StopWatchModeForLibrary stops watching a specific library
func (h *Handler) StopWatchModeForLibrary(libraryID pgtype.UUID) error {
h.mu.Lock()
defer h.mu.Unlock()
libraryIDStr := fmt.Sprintf("%x", libraryID.Bytes)
if !h.watchingLibraries[libraryIDStr] {
return fmt.Errorf("not watching library %s", libraryIDStr)
}
delete(h.watchingLibraries, libraryIDStr)
if len(h.watchingLibraries) == 0 {
h.watchModeCancel()
newCtx, newCancel := context.WithCancel(context.Background())
h.watchModeCtx = newCtx
h.watchModeCancel = newCancel
}
return nil
}
// StartWatchMode handles POST /api/scanner/watch/start
func (h *Handler) StartWatchMode(c *echo.Context) error {
userID := c.Get("user_id").(string)
userUUID, err := uuid.Parse(userID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid user id"})
}
var req struct {
LibraryID string `json:"library_id"`
}
if err := c.Bind(&req); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
if req.LibraryID == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "library_id required"})
}
libraryID, err := uuid.Parse(req.LibraryID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid library id"})
}
if err := h.StartWatchModeForLibrary(c.Request().Context(), pgtype.UUID{Bytes: libraryID, Valid: true}, pgtype.UUID{Bytes: userUUID, Valid: true}); err != nil {
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"message": "watch mode started for library",
"library_id": req.LibraryID,
})
}
// StopWatchMode handles POST /api/scanner/watch/stop
func (h *Handler) StopWatchMode(c *echo.Context) error {
var req struct {
LibraryID string `json:"library_id"`
}
if err := c.Bind(&req); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
if req.LibraryID == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "library_id required"})
}
libraryID, err := uuid.Parse(req.LibraryID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid library id"})
}
if err := h.StopWatchModeForLibrary(pgtype.UUID{Bytes: libraryID, Valid: true}); err != nil {
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"message": "watch mode stopped for library",
"library_id": req.LibraryID,
})
}
// GetWatchModeStatus handles GET /api/scanner/watch/status
func (h *Handler) GetWatchModeStatus(c *echo.Context) error {
h.mu.Lock()
defer h.mu.Unlock()
watchingLibraries := make([]string, 0, len(h.watchingLibraries))
for libID := range h.watchingLibraries {
watchingLibraries = append(watchingLibraries, libID)
}
return c.JSON(http.StatusOK, map[string]interface{}{
"watching_libraries": watchingLibraries,
"total_watching": len(watchingLibraries),
})
}
// StartWatchModeForAllLibraries starts watching all configured libraries
func (h *Handler) StartWatchModeForAllLibraries(ctx context.Context) error {
libraries, err := h.db.ListLibraries(ctx)
if err != nil {
return fmt.Errorf("failed to list libraries: %v", err)
}
for _, library := range libraries {
libraryIDStr := fmt.Sprintf("%x", library.ID.Bytes)
if err := h.StartWatchModeForLibrary(ctx, library.ID, library.CreatedByAdminID); err != nil {
fmt.Printf("Warning: failed to start watch mode for library %s: %v\n", libraryIDStr, err)
continue
}
fmt.Printf("Started watch mode for library %s (%s)\n", library.Name, libraryIDStr)
}
return nil
}