feat(worker): real per-job cancellation for the background job pool

CancelJob existed but was dead code: it only flipped a status in the
in-memory results map - it never cancelled anything, and processJob's
unconditional completion write overwrote the marker anyway.

- derive a per-job context in processJob (parent is the handler ctx
  for manual scans, Background for auto-scans) and register its
  cancel func under the job id; previously every manual scan shared
  the handler's process-lifetime context, so no targeted cancel was
  possible without killing everything
- skip execution at dequeue for jobs cancelled while still queued
- record a pending JobResult on BOTH enqueue paths (Enqueue used by
  auto-scans, EnqueueJob used by the API): queued jobs were invisible
  to status polling (404 until a worker picked them up) and could not
  be cancelled before execution started
- cancelled jobs keep their last real progress instead of a fake 100%
  and drop the meaningless "context canceled" error text
- broadcast the new scan_cancelled websocket event for cancelled scan
  jobs instead of scan_complete, so clients reset their scanning
  indicators rather than reading the abort as a finished scan
- sentinel errors (ErrJobNotFound, ErrJobNotCancellable) so HTTP
  handlers map outcomes to 404/409 without string matching

Unit tests (worker_cancel_test.go) pin: queued-cancel never executes,
running-cancel invokes the context, error mapping, pending recording
on both enqueue paths.

internal/services/worker.go

@@ -4,6 +4,7 @@ import (
 	"bookhoard/internal/database"
 	wsync "bookhoard/internal/sync"
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"path/filepath"
@@ -27,6 +28,13 @@ const (
 	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 (
@@ -45,11 +53,29 @@ 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,
+		}
+	}
+}
+
 type Job struct {
 	ID               string
 	Type             JobType
@@ -148,6 +174,12 @@ type AnalyticsJobResult struct {
 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
@@ -192,6 +224,7 @@ func NewWorkerWithConfig(numWorkers, queueCap int, connManager *wsync.Connection
 	w := &Worker{
 		jobQueue: make(chan *Job, queueCap),
 		results:  make(map[string]*JobResult),
+		cancels:  make(map[string]context.CancelFunc),
 		ctx:      ctx,
 		cancel:   cancel,
 	}
@@ -243,22 +276,40 @@ func (w *Worker) worker() {
 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,
 	}
+
+	// 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
 
-	w.mu.Lock()
-	if result, exists := w.results[job.ID]; exists {
-		result.Status = JobStatusRunning
-	}
-	w.mu.Unlock()
-
 	var err error
 	var result interface{}
 
@@ -318,27 +369,43 @@ func (w *Worker) processJob(job *Job) {
 		}
 	}
 
+	// 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: func() string {
+		Error:     finalError,
-			if err != nil {
-				return err.Error()
-			} else {
-				return ""
-			}
-		}(),
 		Result:    result,
-		Progress:     1.0,
+		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: wsync.MessageTypeScanComplete,
+			Type: msgType,
 			Data: map[string]interface{}{
 				"job_id":        job.ID,
 				"files_scanned": filesScanned,
@@ -972,6 +1039,7 @@ func (w *Worker) EnqueueJob(job *Job) error {
 
 	select {
 	case w.jobQueue <- job:
+		w.recordPending(job)
 		return nil
 	case <-w.ctx.Done():
 		return fmt.Errorf("worker is shutting down")
@@ -1005,19 +1073,33 @@ func (w *Worker) GetActiveScanJobs() []JobResult {
 	return active
 }
 
+// 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()
 
-	if result, exists := w.results[jobID]; exists {
+	result, exists := w.results[jobID]
-		if result.Status == JobStatusRunning || result.Status == JobStatusPending {
+	if !exists {
-			result.Status = JobStatusCancelled
+		return ErrJobNotFound
-			return nil
-		}
-		return fmt.Errorf("job cannot be cancelled")
 	}
 
-	return fmt.Errorf("job not found")
+	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() {

internal/services/worker_cancel_test.go

@@ -0,0 +1,157 @@
+package services
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+// A job cancelled while still queued must never execute. processJob
+// sees the pending-cancel marker at dequeue and leaves the result
+// untouched; a scan without params would otherwise fail with
+// "library_id required" and flip the status to failed — so an empty
+// error field proves the job body never ran.
+func TestCancelQueuedJobNeverExecutes(t *testing.T) {
+	// Zero workers: the job stays queued until processJob is driven by
+	// hand below.
+	w := NewWorkerWithConfig(0, 10, nil)
+
+	job := &Job{
+		ID:      "queued-cancel",
+		Type:    JobTypeScan,
+		Context: context.Background(),
+		Params:  map[string]interface{}{},
+	}
+	if err := w.EnqueueJob(job); err != nil {
+		t.Fatalf("enqueue failed: %v", err)
+	}
+
+	// The pending result must exist before any worker picks the job up,
+	// so status polling can see (and cancel) queued jobs.
+	result, ok := w.GetJobStatus("queued-cancel")
+	if !ok {
+		t.Fatal("queued job has no result entry")
+	}
+	if result.Status != JobStatusPending {
+		t.Fatalf("expected pending status, got %q", result.Status)
+	}
+
+	if err := w.CancelJob("queued-cancel"); err != nil {
+		t.Fatalf("CancelJob failed: %v", err)
+	}
+
+	// Simulate the dequeue that races the cancellation.
+	w.processJob(job)
+
+	result, _ = w.GetJobStatus("queued-cancel")
+	if result.Status != JobStatusCancelled {
+		t.Fatalf("expected cancelled status after dequeue, got %q", result.Status)
+	}
+	if result.Error != "" {
+		t.Errorf("skipped job must carry no processing error, got %q", result.Error)
+	}
+}
+
+// Cancelling a running job must invoke its per-job context cancel func
+// — the old CancelJob only flipped a map status and the job kept
+// running to completion.
+func TestCancelRunningJobCancelsContext(t *testing.T) {
+	w := NewWorker(1, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	job := &Job{ID: "running-cancel", Type: JobTypeScan, Context: ctx}
+
+	// Seed the state processJob establishes while a job runs.
+	w.mu.Lock()
+	w.results[job.ID] = &JobResult{JobID: job.ID, Status: JobStatusRunning}
+	w.cancels[job.ID] = cancel
+	w.mu.Unlock()
+
+	if err := w.CancelJob(job.ID); err != nil {
+		t.Fatalf("CancelJob failed: %v", err)
+	}
+
+	if ctx.Err() == nil {
+		t.Fatal("CancelJob must cancel the job's context")
+	}
+	result, _ := w.GetJobStatus(job.ID)
+	if result.Status != JobStatusCancelled {
+		t.Fatalf("expected cancelled status, got %q", result.Status)
+	}
+}
+
+func TestCancelJobErrors(t *testing.T) {
+	w := NewWorker(1, nil)
+
+	if err := w.CancelJob("missing-job"); !errors.Is(err, ErrJobNotFound) {
+		t.Errorf("unknown job: expected ErrJobNotFound, got %v", err)
+	}
+
+	w.mu.Lock()
+	w.results["done-job"] = &JobResult{JobID: "done-job", Status: JobStatusCompleted}
+	w.results["failed-job"] = &JobResult{JobID: "failed-job", Status: JobStatusFailed}
+	w.results["cancelled-job"] = &JobResult{JobID: "cancelled-job", Status: JobStatusCancelled}
+	w.mu.Unlock()
+
+	for _, id := range []string{"done-job", "failed-job", "cancelled-job"} {
+		if err := w.CancelJob(id); !errors.Is(err, ErrJobNotCancellable) {
+			t.Errorf("terminal job %q: expected ErrJobNotCancellable, got %v", id, err)
+		}
+	}
+}
+
+// Both enqueue paths must register a pending result: auto-scans arrive
+// via Enqueue (fire-and-forget) and manual scans via EnqueueJob, and a
+// queued job must be visible to status polling and cancellable before
+// a worker dequeues it.
+func TestEnqueueRecordsPendingResult(t *testing.T) {
+	w := NewWorkerWithConfig(0, 10, nil)
+
+	fireAndForget := &Job{
+		ID:      "via-enqueue",
+		Type:    JobTypeScan,
+		Context: context.Background(),
+		Params:  map[string]interface{}{"library_id": "lib-auto"},
+	}
+	w.Enqueue(fireAndForget)
+
+	tracked := &Job{
+		ID:      "via-enqueue-job",
+		Type:    JobTypeScan,
+		Context: context.Background(),
+		Params:  map[string]interface{}{"library_id": "lib-manual"},
+	}
+	if err := w.EnqueueJob(tracked); err != nil {
+		t.Fatalf("EnqueueJob failed: %v", err)
+	}
+
+	for id, libID := range map[string]string{
+		"via-enqueue":     "lib-auto",
+		"via-enqueue-job": "lib-manual",
+	} {
+		result, ok := w.GetJobStatus(id)
+		if !ok {
+			t.Fatalf("%s: no pending result recorded", id)
+		}
+		if result.Status != JobStatusPending {
+			t.Errorf("%s: expected pending, got %q", id, result.Status)
+		}
+		if result.LibraryID != libID {
+			t.Errorf("%s: expected library_id %q, got %q", id, libID, result.LibraryID)
+		}
+	}
+}
+
+// A scan whose context is already cancelled must abort immediately
+// with context.Canceled — before counting, walking, or running the
+// archive pass. processJob classifies that error as a cancelled job.
+func TestScanFoldersCancelledContext(t *testing.T) {
+	s := NewMediaScanner(nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	if err := s.ScanFolders(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+}

internal/sync/websocket.go

@@ -20,6 +20,7 @@ const (
 	MessageTypeScanProgress     = "scan_progress"
 	MessageTypeScanComplete     = "scan_complete"
 	MessageTypeScanError        = "scan_error"
+	MessageTypeScanCancelled    = "scan_cancelled"
 )
 
 // BroadcastMessage represents a message to broadcast to connected clients