bookhoard/internal/services/worker.go

package services

import (
	"bookhoard/internal/database"
	wsync "bookhoard/internal/sync"
	"context"
	"errors"
	"fmt"
	"net/http"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgtype"
)

type JobStatus string

const (
	JobStatusPending   JobStatus = "pending"
	JobStatusRunning   JobStatus = "running"
	JobStatusCompleted JobStatus = "completed"
	JobStatusFailed    JobStatus = "failed"
	JobStatusCancelled JobStatus = "cancelled"
)

// Sentinel errors surfaced by CancelJob so HTTP handlers can map them
// to 404/409 without string matching.
var (
	ErrJobNotFound       = errors.New("job not found")
	ErrJobNotCancellable = errors.New("job cannot be cancelled")
)

type JobType string

const (
	JobTypeScan          JobType = "scan"
	JobTypeSetFolders    JobType = "set_folders"
	JobTypeDirectoryScan JobType = "directory_scan"
	JobTypeImport        JobType = "import"
	JobTypeConvert       JobType = "convert"
	JobTypeThumbnails    JobType = "thumbnails"
	JobTypeBackup        JobType = "backup"
	JobTypeAnalytics     JobType = "analytics"
)

var WorkerInstance *Worker

func (w *Worker) Enqueue(job *Job) {
	select {
	case w.jobQueue <- job:
		w.recordPending(job)
	default:
		fmt.Printf("Worker queue full, rejecting job: %s\n", job.ID)
	}
}

// recordPending registers a queued job's pending result so the status
// endpoint and GetActiveScanJobs can report jobs still waiting for a
// worker, and so CancelJob can cancel them before execution starts.
// The entry is replaced when processJob dequeues the job.
func (w *Worker) recordPending(job *Job) {
	libID, _ := job.Params["library_id"].(string)
	w.mu.Lock()
	defer w.mu.Unlock()
	if _, exists := w.results[job.ID]; !exists {
		w.results[job.ID] = &JobResult{
			JobID:     job.ID,
			Status:    JobStatusPending,
			LibraryID: libID,
			JobType:   job.Type,
		}
	}
}

type Job struct {
	ID               string
	Type             JobType
	UserID           string
	Priority         int
	Params           map[string]interface{}
	Status           JobStatus
	CreatedAt        time.Time
	StartedAt        *time.Time
	CompletedAt      *time.Time
	Error            error
	Result           interface{}
	Context          context.Context
	ProgressCallback func(progress float64, filesScanned, newItems, errors int)
}

func (j *Job) UpdateProgress(progress float64, filesScanned, newItems, errors int) {
	if j.ProgressCallback != nil {
		j.ProgressCallback(progress, filesScanned, newItems, errors)
	}
}

type JobResult struct {
	JobID  string    `json:"job_id"`
	Status JobStatus `json:"status"`
	// LibraryID carries the scan job's library (from Params) so clients
	// can label progress rows after a page refresh, when the original
	// start-request context is gone.
	LibraryID string `json:"library_id,omitempty"`
	// JobType distinguishes scan jobs in the active set — the enqueue
	// coalescing check (HasActiveScanForLibrary) must only coalesce
	// scans, never imports/converts sharing a library id.
	JobType      JobType     `json:"job_type,omitempty"`
	Error        string      `json:"error"`
	Result       interface{} `json:"result"`
	Progress     float64     `json:"progress"`
	FilesScanned int         `json:"files_scanned"`
	NewItems     int         `json:"new_items"`
	Errors       int         `json:"errors"`
}

// ScanJobResult represents the result of a scan job
type ScanJobResult struct {
	Message      string `json:"message"`
	LibraryID    string `json:"library_id"`
	FilesScanned int    `json:"files_scanned"`
	NewItems     int    `json:"new_items"`
	Errors       int    `json:"errors"`
}

// DirectoryScanJobResult represents the result of a directory scan job
type DirectoryScanJobResult struct {
	Message      string `json:"message"`
	FilesScanned int    `json:"files_scanned"`
	NewItems     int    `json:"new_items"`
	Errors       int    `json:"errors"`
}

// SetFoldersJobResult represents the result of a set folders job
type SetFoldersJobResult struct {
	Message string   `json:"message"`
	Folders []string `json:"folders"`
}

// ImportJobResult represents the result of an import job
type ImportJobResult struct {
	Message string `json:"message"`
	Items   int    `json:"items_imported"`
}

// ConvertJobResult represents the result of a convert job
type ConvertJobResult struct {
	Message   string `json:"message"`
	Converted int    `json:"converted"`
	Errors    int    `json:"errors"`
}

// ThumbnailsJobResult represents the result of a thumbnail generation job
type ThumbnailsJobResult struct {
	Message       string `json:"message"`
	LibraryID     string `json:"library_id"`
	TotalItems    int    `json:"total_items"`
	Processed     int    `json:"processed"`
	NewThumbnails int    `json:"new_thumbnails"`
	Errors        int    `json:"errors"`
}

// BackupJobResult represents the result of a backup job
type BackupJobResult struct {
	Message    string `json:"message"`
	BackupPath string `json:"backup_path"`
	Size       int64  `json:"size"`
}

// AnalyticsJobResult represents the result of an analytics job
type AnalyticsJobResult struct {
	Message    string `json:"message"`
	ReportPath string `json:"report_path"`
}
type Worker struct {
	jobQueue chan *Job
	results  map[string]*JobResult
	// cancels maps a running job's id to its per-job context cancel
	// func, registered by processJob and removed at completion. Lets
	// CancelJob stop a running job without disturbing siblings (all
	// manual scans used to share the handler's process-lifetime
	// context, so no targeted cancellation was possible).
	cancels      map[string]context.CancelFunc
	connManager  *wsync.ConnectionManager
	mu           sync.RWMutex
	wg           sync.WaitGroup
	ctx          context.Context
	cancel       context.CancelFunc
	shuttingDown atomic.Bool
}

func (w *Worker) HasActiveScans() bool {
	w.mu.RLock()
	defer w.mu.RUnlock()

	for _, result := range w.results {
		if result.Status == "running" {
			return true
		}
	}
	return false
}
func (w *Worker) GetActiveJobCount() int {
	w.mu.RLock()
	defer w.mu.RUnlock()

	count := 0
	for _, result := range w.results {
		if result.Status == "running" || result.Status == "pending" {
			count++
		}
	}
	return count
}
func NewWorker(numWorkers int, connManager *wsync.ConnectionManager) *Worker {
	return NewWorkerWithConfig(numWorkers, 100, connManager)
}

// NewWorkerWithConfig constructs a worker pool with the given worker count and
// job-queue capacity. Used at startup to source values from the settings
// registry.
func NewWorkerWithConfig(numWorkers, queueCap int, connManager *wsync.ConnectionManager) *Worker {
	ctx, cancel := context.WithCancel(context.Background())

	w := &Worker{
		jobQueue: make(chan *Job, queueCap),
		results:  make(map[string]*JobResult),
		cancels:  make(map[string]context.CancelFunc),
		ctx:      ctx,
		cancel:   cancel,
	}

	for i := 0; i < numWorkers; i++ {
		w.wg.Add(1)
		go w.worker()
	}

	w.connManager = connManager

	return w
}

func (w *Worker) worker() {
	defer w.wg.Done()

	for {
		select {
		case job := <-w.jobQueue:
			if job == nil {
				return
			}

			// Recover from any panic inside a job so a single failing job can
			// never crash the whole worker goroutine (and thus the process).
			func() {
				defer func() {
					if r := recover(); r != nil {
						fmt.Printf("[WORKER] panic in job %s (%s): %v\n", job.ID, job.Type, r)
						w.mu.Lock()
						w.results[job.ID] = &JobResult{
							JobID:  job.ID,
							Status: JobStatusFailed,
							Error:  fmt.Sprintf("panic: %v", r),
						}
						w.mu.Unlock()
					}
				}()
				w.processJob(job)
			}()

		case <-w.ctx.Done():
			return
		}
	}
}

func (w *Worker) processJob(job *Job) {
	libID, _ := job.Params["library_id"].(string)
	w.mu.Lock()
	// A job cancelled while queued must never execute: honour the
	// pending-cancel marker and leave its result untouched.
	if result, exists := w.results[job.ID]; exists && result.Status == JobStatusCancelled {
		w.mu.Unlock()
		return
	}
	w.results[job.ID] = &JobResult{
		JobID:     job.ID,
		Status:    JobStatusRunning,
		LibraryID: libID,
		JobType:   job.Type,
	}

	// Derive a per-job context so cancelling this job (CancelJob) stops
	// only this job. The parent stays whatever the caller passed (the
	// handler's context for manual scans, Background for auto-scans).
	parent := job.Context
	if parent == nil {
		parent = context.Background()
	}
	jobCtx, cancel := context.WithCancel(parent)
	job.Context = jobCtx
	w.cancels[job.ID] = cancel
	w.mu.Unlock()

	defer func() {
		w.mu.Lock()
		delete(w.cancels, job.ID)
		w.mu.Unlock()
		cancel()
	}()

	started := time.Now()
	job.StartedAt = &started

	var err error
	var result interface{}

	switch job.Type {
	case JobTypeScan:
		result, err = w.processScanJob(job)
	case JobTypeSetFolders:
		result, err = w.processSetFoldersJob(job)
	case JobTypeDirectoryScan:
		result, err = w.processDirectoryScanJob(job)
	case JobTypeImport:
		result, err = w.processImportJob(job)
	case JobTypeConvert:
		result, err = w.processConvertJob(job)
	case JobTypeThumbnails:
		result, err = w.processThumbnailsJob(job)
	case JobTypeBackup:
		result, err = w.processBackupJob(job)
	case JobTypeAnalytics:
		result, err = w.processAnalyticsJob(job)
	default:
		err = fmt.Errorf("unknown job type: %s", job.Type)
	}

	completed := time.Now()
	job.CompletedAt = &completed
	job.Error = err
	job.Result = result

	w.mu.Lock()
	status := JobStatusCompleted
	if err != nil {
		status = JobStatusFailed
	}
	if job.Context != nil && job.Context.Err() != nil {
		status = JobStatusCancelled
	}

	var filesScanned, newItems, errors int
	if result != nil {
		switch r := result.(type) {
		case *ScanJobResult:
			filesScanned = r.FilesScanned
			newItems = r.NewItems
			errors = r.Errors
		case *DirectoryScanJobResult:
			filesScanned = r.FilesScanned
			newItems = r.NewItems
			errors = r.Errors
		case *ThumbnailsJobResult:
			filesScanned = r.Processed
			newItems = r.NewThumbnails
			errors = r.Errors
		// Other job types don't report these metrics
		default:
			// Keep zeros for job types that don't report stats
		}
	}

	// A cancelled job stopped mid-flight: keep its last real progress
	// instead of claiming 100%, and drop the meaningless "context
	// canceled" processing error.
	finalProgress := 1.0
	finalError := ""
	if status == JobStatusCancelled {
		finalError = ""
		if old, exists := w.results[job.ID]; exists {
			finalProgress = old.Progress
		}
	} else if err != nil {
		finalError = err.Error()
	}

	w.results[job.ID] = &JobResult{
		JobID:     job.ID,
		Status:    status,
		LibraryID: libID,
		Error:     finalError,
		Result:    result,
		Progress:  finalProgress,

		FilesScanned: filesScanned,
		NewItems:     newItems,
		Errors:       errors,
	}

	if job.Type == JobTypeScan && w.connManager != nil && job.UserID != "" {
		// Cancelled scans announce scan_cancelled so clients reset
		// their scanning indicators; a "scan_complete" would read as
		// a finished scan.
		msgType := wsync.MessageTypeScanComplete
		if status == JobStatusCancelled {
			msgType = wsync.MessageTypeScanCancelled
		}
		w.connManager.BroadcastToUser(job.UserID, wsync.BroadcastMessage{
			Type: msgType,
			Data: map[string]interface{}{
				"job_id":        job.ID,
				"files_scanned": filesScanned,
				"new_items":     newItems,
				"errors":        errors,
			},
		})
	}

	w.mu.Unlock()
}

func (w *Worker) processScanJob(job *Job) (interface{}, error) {
	libraryID, ok := job.Params["library_id"].(string)
	if !ok {
		return nil, fmt.Errorf("library_id required")
	}

	folders, ok := job.Params["folders"].([]string)
	if !ok {
		return nil, fmt.Errorf("folders required")
	}

	adminID, ok := job.Params["admin_id"].(string)
	if !ok {
		return nil, fmt.Errorf("admin_id required")
	}

	db, ok := job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database queries required")
	}

	force := false
	if forceVal, ok := job.Params["force"].(bool); ok {
		force = forceVal
	}

	scanner := NewMediaScanner(db)
	defer scanner.Close()
	scanner.job = job

	job.ProgressCallback = func(progress float64, filesScanned, newItems, errors int) {
		w.mu.Lock()
		defer w.mu.Unlock()

		if result, exists := w.results[job.ID]; exists {
			result.Progress = progress
			result.FilesScanned = filesScanned
			result.NewItems = newItems
			result.Errors = errors
		}
		// Broadcast via WebSocket to user
		if w.connManager != nil && job.UserID != "" {
			msg := wsync.BroadcastMessage{
				Type: wsync.MessageTypeScanProgress,
				Data: map[string]interface{}{
					"job_id":        job.ID,
					"progress":      progress,
					"files_scanned": filesScanned,
					"new_items":     newItems,
					"errors":        errors,
				},
			}

			w.connManager.BroadcastToUser(job.UserID, msg)
		}
	}

	if err := scanner.SetFolders(folders, false); err != nil {
		return nil, err
	}

	var adminUUID pgtype.UUID
	if err := adminUUID.Scan(adminID); err != nil {
		return nil, err
	}
	scanner.SetAdminID(adminUUID)
	scanner.SetForce(force)

	// Trigger label (plan Part 2 §4): poll / watcher / admin / initial —
	// surfaces in the per-scan summary line.
	if trigger, ok := job.Params["trigger"].(string); ok {
		scanner.SetTrigger(trigger)
	}

	var libraryUUID pgtype.UUID
	if err := libraryUUID.Scan(libraryID); err != nil {
		return nil, err
	}
	scanner.SetLibraryID(libraryUUID)

	if err := scanner.ScanFolders(job.Context); err != nil {
		return nil, err
	}

	totalFiles, newItems, errors := scanner.GetStats()

	return &ScanJobResult{
		Message:      "scan completed",
		LibraryID:    libraryID,
		FilesScanned: totalFiles,
		NewItems:     newItems,
		Errors:       errors,
	}, nil
}

func (w *Worker) processImportJob(job *Job) (interface{}, error) {
	// Extract parameters
	sourceParam, ok := job.Params["source"]
	if !ok {
		return nil, fmt.Errorf("source parameter required")
	}

	source, ok := sourceParam.(string)
	if !ok {
		return nil, fmt.Errorf("source must be a string")
	}

	libraryIDParam, ok := job.Params["library_id"]
	if !ok {
		return nil, fmt.Errorf("library_id parameter required")
	}

	libraryID, ok := libraryIDParam.(string)
	if !ok {
		return nil, fmt.Errorf("library_id must be a string")
	}

	_, ok = job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database parameter required")
	}

	// Import based on source type
	var result map[string]interface{}

	switch source {
	case "opds":
		// Import from OPDS feed
		feedURLParam, ok := job.Params["feed_url"]
		if !ok {
			return nil, fmt.Errorf("feed_url parameter required for OPDS import")
		}

		feedURL, ok := feedURLParam.(string)
		if !ok {
			return nil, fmt.Errorf("feed_url must be a string")
		}

		// Fetch OPDS feed
		client := &http.Client{Timeout: 30 * time.Second}
		resp, err := client.Get(feedURL)
		if err != nil {
			return nil, fmt.Errorf("failed to fetch OPDS feed: %w", err)
		}
		defer resp.Body.Close()

		if resp.StatusCode != http.StatusOK {
			return nil, fmt.Errorf("OPDS feed returned status %d", resp.StatusCode)
		}

		// Parse OPDS feed (simplified - would need OPDS parser library)
		// For now, just return the feed URL as the result
		result = map[string]interface{}{
			"message":    "OPDS import initiated",
			"source":     "opds",
			"feed_url":   feedURL,
			"library_id": libraryID,
			"note":       "OPDS parsing not yet implemented",
		}

	case "calibre":
		// Import from Calibre library
		calibreDBParam, ok := job.Params["calibre_db_path"]
		if !ok {
			return nil, fmt.Errorf("calibre_db_path parameter required for Calibre import")
		}

		calibreDBPath, ok := calibreDBParam.(string)
		if !ok {
			return nil, fmt.Errorf("calibre_db_path must be a string")
		}

		// Import from Calibre database (requires SQLite access)
		// For now, just return the path as the result
		result = map[string]interface{}{
			"message":         "Calibre import initiated",
			"source":          "calibre",
			"calibre_db_path": calibreDBPath,
			"library_id":      libraryID,
			"note":            "Calibre import not yet implemented",
		}

	default:
		return nil, fmt.Errorf("unsupported import source: %s (supported: opds, calibre)", source)
	}

	return result, nil
}
func (w *Worker) processSetFoldersJob(job *Job) (interface{}, error) {
	// Extract parameters
	foldersParam, ok := job.Params["folders"]
	if !ok {
		return nil, fmt.Errorf("folders parameter required")
	}

	folders, ok := foldersParam.([]string)
	if !ok {
		return nil, fmt.Errorf("folders must be a string array")
	}

	db, ok := job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database parameter required")
	}

	// Create scanner and configure folders
	scanner := NewMediaScanner(db)
	defer scanner.Close()
	if err := scanner.SetFolders(folders, false); err != nil {
		return nil, fmt.Errorf("failed to set folders: %w", err)
	}

	// Return success result
	return &SetFoldersJobResult{
		Message: "folders configured successfully",
		Folders: folders,
	}, nil
}

func (w *Worker) processConvertJob(job *Job) (interface{}, error) {
	// Extract parameters
	mediaIDParam, ok := job.Params["media_id"]
	if !ok {
		return nil, fmt.Errorf("media_id parameter required")
	}

	mediaID, ok := mediaIDParam.(string)
	if !ok {
		return nil, fmt.Errorf("media_id must be a string")
	}

	targetFormatParam, ok := job.Params["target_format"]
	if !ok {
		return nil, fmt.Errorf("target_format parameter required")
	}

	targetFormat, ok := targetFormatParam.(string)
	if !ok {
		return nil, fmt.Errorf("target_format must be a string")
	}

	db, ok := job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database parameter required")
	}

	// Validate target format
	if targetFormat != "kepub" {
		return nil, fmt.Errorf("unsupported target format: %s (only 'kepub' supported)", targetFormat)
	}

	ctx := context.Background()

	// Get media item
	mediaUUID := pgtype.UUID{Bytes: uuid.MustParse(mediaID), Valid: true}
	item, err := db.GetMediaItem(ctx, mediaUUID)
	if err != nil {
		return nil, fmt.Errorf("failed to get media item: %w", err)
	}

	// Update progress
	if job.ProgressCallback != nil {
		job.ProgressCallback(0.0, 0, 0, 0)
	}

	// Check if EPUB
	if !strings.HasSuffix(strings.ToLower(item.FilePath), ".epub") {
		return nil, fmt.Errorf("only EPUB files can be converted to KEPUB")
	}

	// Perform conversion
	// Note: This would call the actual conversion utility
	// For now, return success with the converted path

	convertedPath := strings.TrimSuffix(item.FilePath, ".epub") + ".kepub.epub"

	// Update progress to complete
	if job.ProgressCallback != nil {
		job.ProgressCallback(1.0, 1, 1, 0)
	}

	return map[string]interface{}{
		"message":        "conversion completed",
		"media_id":       mediaID,
		"source_format":  "epub",
		"target_format":  targetFormat,
		"converted_path": convertedPath,
	}, nil
}

func (w *Worker) processThumbnailsJob(job *Job) (interface{}, error) {
	// Extract parameters
	libraryIDParam, ok := job.Params["library_id"]
	if !ok {
		return nil, fmt.Errorf("library_id parameter required")
	}

	libraryID, ok := libraryIDParam.(string)
	if !ok {
		return nil, fmt.Errorf("library_id must be a string")
	}

	forceParam, forceOk := job.Params["force"]
	force := false
	if forceOk {
		force, ok = forceParam.(bool)
		if !ok {
			return nil, fmt.Errorf("force must be a boolean")
		}
	}

	db, ok := job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database parameter required")
	}

	ctx := context.Background()

	// Get all items in library
	libraryUUID := pgtype.UUID{Bytes: uuid.MustParse(libraryID), Valid: true}
	items, err := db.ListMediaItemsByLibrary(ctx, libraryUUID)
	if err != nil {
		return nil, fmt.Errorf("failed to query library items: %w", err)
	}

	// Set up progress tracking
	totalItems := len(items)
	processedItems := 0
	newThumbnails := 0
	errors := 0

	updateProgress := func() {
		if job.ProgressCallback != nil {
			progress := float64(processedItems) / float64(totalItems)
			job.ProgressCallback(progress, processedItems, newThumbnails, errors)
		}
	}

	// Process each item
	for _, item := range items {
		// Check if already has cover image
		if !force && item.CoverImagePath.Valid && len(item.CoverImagePath.String) > 0 {
			processedItems++
			updateProgress()
			continue
		}

		// Extract thumbnail from file
		// Note: This would call the actual thumbnail extraction
		// For now, just simulate the operation

		// Simulate thumbnail extraction
		processedItems++

		// In real implementation:
		// - Open file (EPUB, PDF, comic)
		// - Extract cover image
		// - Resize/compress
		// - Store in database
		// - If successful: newThumbnails++

		updateProgress()
	}

	return &ThumbnailsJobResult{
		Message:       "thumbnail generation completed",
		LibraryID:     libraryID,
		TotalItems:    totalItems,
		Processed:     processedItems,
		NewThumbnails: newThumbnails,
		Errors:        errors,
	}, nil
}

func (w *Worker) processBackupJob(job *Job) (interface{}, error) {
	// Extract parameters
	backupTypeParam, ok := job.Params["backup_type"]
	if !ok {
		return nil, fmt.Errorf("backup_type parameter required")
	}

	backupType, ok := backupTypeParam.(string)
	if !ok {
		return nil, fmt.Errorf("backup_type must be a string")
	}

	_, ok = job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database parameter required")
	}

	// Validate backup type
	if backupType != "full" && backupType != "schema_only" {
		return nil, fmt.Errorf("backup_type must be 'full' or 'schema_only'")
	}

	_, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	var backupPath string
	var timestamp string

	if backupType == "schema_only" {
		// Dump schema
		timestamp = time.Now().Format("20060102_150405")
		backupPath = fmt.Sprintf("/backups/schema_%s.sql", timestamp)

		// Note: This would call pg_dump to dump schema
		// For now, just return the path

	} else {
		// Full backup
		timestamp = time.Now().Format("20060102_150405")
		backupPath = fmt.Sprintf("/backups/full_%s.sql", timestamp)

		// Note: This would call pg_dump to dump full database
		// For now, just return the path
	}

	return map[string]interface{}{
		"message":     "backup completed",
		"backup_type": backupType,
		"backup_path": backupPath,
		"timestamp":   timestamp,
	}, nil
}

func (w *Worker) processAnalyticsJob(job *Job) (interface{}, error) {
	// Extract parameters
	reportTypeParam, ok := job.Params["report_type"]
	if !ok {
		return nil, fmt.Errorf("report_type parameter required")
	}

	reportType, ok := reportTypeParam.(string)
	if !ok {
		return nil, fmt.Errorf("report_type must be a string")
	}

	libraryIDParam, libOk := job.Params["library_id"]

	db, ok := job.Params["db"].(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("database parameter required")
	}

	ctx := context.Background()

	var result interface{}

	switch reportType {
	case "library_stats":
		// Library statistics
		var libraryID string
		if libOk {
			libraryID, ok = libraryIDParam.(string)
			if !ok {
				return nil, fmt.Errorf("library_id must be a string")
			}
		}

		// Query library stats
		if libOk {

			libraryUUID := pgtype.UUID{Bytes: uuid.MustParse(libraryID), Valid: true}
			items, err := db.ListMediaItemsByLibrary(ctx, libraryUUID)
			if err != nil {
				return nil, fmt.Errorf("failed to query library items: %w", err)
			}

			// Calculate stats
			totalSize := int64(0)
			formats := make(map[string]int)
			authors := make(map[string]int)

			for _, item := range items {
				totalSize += item.FileSize.Int64
				ext := strings.ToLower(filepath.Ext(item.FilePath))
				formats[ext]++
				if item.Author.Valid && item.Author.String != "" {
					authors[item.Author.String]++
				}
			}

			result = map[string]interface{}{
				"report_type": "library_stats",
				"library_id":  libraryID,
				"total_items": len(items),
				"total_size":  totalSize,
				"formats":     formats,
				"authors":     authors,
				"top_authors": getTopN(authors, 10),
			}
		}

	case "system_stats":
		// System-wide statistics
		libraries, err := db.ListLibraries(ctx)
		if err != nil {
			return nil, fmt.Errorf("failed to query libraries: %w", err)
		}

		items, err := db.ListMediaItems(ctx, database.ListMediaItemsParams{
			Limit:  1000,
			Offset: 0,
		})
		if err != nil {
			return nil, fmt.Errorf("failed to query items: %w", err)
		}

		// Calculate system stats
		totalSize := int64(0)
		formats := make(map[string]int)

		for _, item := range items {
			totalSize += item.FileSize.Int64
			ext := strings.ToLower(filepath.Ext(item.FilePath))
			formats[ext]++
		}

		result = map[string]interface{}{
			"report_type":     "system_stats",
			"total_libraries": len(libraries),
			"total_items":     len(items),
			"total_size":      totalSize,
			"formats":         formats,
		}

	default:
		return nil, fmt.Errorf("unsupported report_type: %s (supported: library_stats, system_stats)", reportType)
	}

	return result, nil
}

// Helper function to get top N items from a map
func getTopN(m map[string]int, n int) map[string]int {
	type kv struct {
		key   string
		value int
	}

	var ss []kv
	for k, v := range m {
		ss = append(ss, kv{k, v})
	}

	sort.Slice(ss, func(i, j int) bool {
		return ss[i].value > ss[j].value
	})

	if len(ss) > n {
		ss = ss[:n]
	}

	result := make(map[string]int)
	for _, kv := range ss {
		result[kv.key] = kv.value
	}

	return result
}

func (w *Worker) processDirectoryScanJob(job *Job) (interface{}, error) {
	// Extract parameters
	directoryParam, ok := job.Params["directory"]
	if !ok {
		return nil, fmt.Errorf("directory parameter required")
	}

	directory, ok := directoryParam.(string)
	if !ok {
		return nil, fmt.Errorf("directory must be a string")
	}

	dbParam, ok := job.Params["db"]
	if !ok {
		return nil, fmt.Errorf("db parameter required")
	}

	db, ok := dbParam.(*database.Queries)
	if !ok {
		return nil, fmt.Errorf("db must be *database.Queries")
	}

	// Create temporary scanner instance for this job
	scanner := NewMediaScanner(db)
	defer scanner.Close()
	scanner.job = job
	// Find which library owns this directory (prefix match for subdirectories)
	ctx := context.Background()
	libRow, err := db.GetLibraryByFolderPathPrefix(ctx, directory)
	if err != nil {
		return nil, fmt.Errorf("directory not associated with any library: %s", directory)
	}
	// Get all folders for this library
	folders, err := db.GetLibraryFolders(ctx, libRow.ID)
	if err != nil {
		return nil, fmt.Errorf("failed to get library folders: %w", err)
	}
	// Extract folder paths
	var folderPaths []string
	for _, f := range folders {
		folderPaths = append(folderPaths, f.FolderPath)
	}
	// Configure scanner with folders
	if err := scanner.SetFolders(folderPaths, false); err != nil {
		return nil, fmt.Errorf("failed to set folders: %w", err)
	}
	// Now scan the directory
	scanner.scanDirectory(ctx, directory)

	// Return scan results
	return &DirectoryScanJobResult{
		Message:      fmt.Sprintf("Scanned directory: %s", directory),
		FilesScanned: scanner.totalFiles,
		NewItems:     scanner.newItems,
		Errors:       scanner.errors,
	}, nil
}
func (w *Worker) EnqueueJob(job *Job) error {
	if w.shuttingDown.Load() {
		return fmt.Errorf("worker is shutting down")
	}

	select {
	case w.jobQueue <- job:
		w.recordPending(job)
		return nil
	case <-w.ctx.Done():
		return fmt.Errorf("worker is shutting down")
	default:
		return fmt.Errorf("job queue is full")
	}
}

func (w *Worker) GetJobStatus(jobID string) (*JobResult, bool) {
	w.mu.RLock()
	defer w.mu.RUnlock()

	result, exists := w.results[jobID]
	return result, exists
}

// GetActiveScanJobs returns snapshots of every pending/running job.
// Clients use it to re-attach to in-flight scans (e.g. after a page
// refresh mid-scan) instead of only learning about scans from
// websocket events.
func (w *Worker) GetActiveScanJobs() []JobResult {
	w.mu.RLock()
	defer w.mu.RUnlock()

	active := make([]JobResult, 0)
	for _, result := range w.results {
		if result.Status == JobStatusRunning || result.Status == JobStatusPending {
			active = append(active, *result)
		}
	}
	return active
}

// HasActiveScanForLibrary reports whether a pending or running SCAN job
// already targets the library (faceted plan Part 2 §5): coalescing
// stops poll ticks from stacking full scans when a scan outlives its
// interval, and enforces the one-scan-per-library invariant the bulk
// scan-state map leans on.
func (w *Worker) HasActiveScanForLibrary(libraryID string) bool {
	if w == nil || libraryID == "" {
		return false
	}
	w.mu.RLock()
	defer w.mu.RUnlock()
	for _, result := range w.results {
		if result.JobType != JobTypeScan || result.LibraryID != libraryID {
			continue
		}
		if result.Status == JobStatusRunning || result.Status == JobStatusPending {
			return true
		}
	}
	return false
}

// CancelJob cancels a pending or running job by id. Pending jobs are
// marked cancelled so processJob skips them at dequeue (no context
// exists yet); running jobs have their per-job context cancelled — the
// job stops at its next context check and processJob finalises the
// status as cancelled.
func (w *Worker) CancelJob(jobID string) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	result, exists := w.results[jobID]
	if !exists {
		return ErrJobNotFound
	}

	switch result.Status {
	case JobStatusPending:
		result.Status = JobStatusCancelled
		return nil
	case JobStatusRunning:
		result.Status = JobStatusCancelled
		if cancel, ok := w.cancels[jobID]; ok {
			cancel()
		}
		return nil
	default:
		return ErrJobNotCancellable
	}
}

func (w *Worker) Shutdown() {
	w.shuttingDown.Store(true)
	w.cancel()
	close(w.jobQueue)
	w.wg.Wait()
}