john-okeefe
40f303b004
- Add UserID field to Job struct for tracking job ownership - Broadcast scan progress updates to user's WebSocket connections - Send real-time updates during scanning (progress, files scanned, new items, errors) This allows the frontend to display live scan progress without HTTP polling. Scanner now associates scan jobs with requesting user for targeted updates.
336 lines
10 KiB
Go
336 lines
10 KiB
Go
package handlers
|
|
|
|
import (
|
|
"bookhoard/internal/services"
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/labstack/echo/v4"
|
|
)
|
|
|
|
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: [16]byte(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"})
|
|
}
|
|
|
|
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,
|
|
},
|
|
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
|
|
if err := h.scanner.SetFolders(req.FolderPaths); 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,
|
|
})
|
|
}
|
|
|
|
// 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); 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: [16]byte(libraryID), Valid: true}, pgtype.UUID{Bytes: [16]byte(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: [16]byte(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
|
|
}
|