bookhoard/internal/services/worker.go

package services

import (
	"bookhoard/internal/database"
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"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"
)

type JobType string

const (
	JobTypeScan JobType = "scan"
)

type Job struct {
	ID               string
	Type             JobType
	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
	Status       JobStatus
	Error        string
	Result       interface{}
	Progress     float64
	FilesScanned int
	NewItems     int
	Errors       int
}

type Worker struct {
	jobQueue     chan *Job
	results      map[string]*JobResult
	mu           sync.RWMutex
	wg           sync.WaitGroup
	ctx          context.Context
	cancel       context.CancelFunc
	shuttingDown atomic.Bool
}

func NewWorker(numWorkers int) *Worker {
	ctx, cancel := context.WithCancel(context.Background())

	w := &Worker{
		jobQueue: make(chan *Job, 100),
		results:  make(map[string]*JobResult),
		ctx:      ctx,
		cancel:   cancel,
	}

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

	return w
}

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

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

			w.processJob(job)

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

func (w *Worker) processJob(job *Job) {
	w.mu.Lock()
	w.results[job.ID] = &JobResult{
		JobID:  job.ID,
		Status: JobStatusRunning,
	}
	w.mu.Unlock()

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

	w.mu.Lock()
	if result, exists := w.results[job.ID]; exists {
		result.Status = JobStatusRunning
	}
	w.mu.Unlock()

	var err error
	var result interface{}

	switch job.Type {
	case JobTypeScan:
		result, err = w.processScanJob(job)
	default:
		err = fmt.Errorf("unknown job type: %s", job.Type)
	}

	completedAt := time.Now()
	job.CompletedAt = &completedAt
	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 {
		if stats, ok := result.(map[string]interface{}); ok {
			filesScanned = int(stats["files_scanned"].(float64))
			newItems = int(stats["new_items"].(float64))
			errors = int(stats["errors"].(float64))
		}
	}

	w.results[job.ID] = &JobResult{
		JobID:  job.ID,
		Status: status,
		Error: func() string {
			if err != nil {
				return err.Error()
			} else {
				return ""
			}
		}(),
		Result:       result,
		Progress:     1.0,
		FilesScanned: filesScanned,
		NewItems:     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")
	}

	scanner := NewMediaScanner(db)
	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
		}
	}

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

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

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

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

	return map[string]interface{}{
		"message":       "scan completed",
		"library_id":    libraryID,
		"files_scanned": totalFiles,
		"new_items":     newItems,
		"errors":        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:
		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
}

func (w *Worker) CancelJob(jobID string) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	if result, exists := w.results[jobID]; exists {
		if result.Status == JobStatusRunning || result.Status == JobStatusPending {
			result.Status = JobStatusCancelled
			return nil
		}
		return fmt.Errorf("job cannot be cancelled")
	}

	return fmt.Errorf("job not found")
}

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