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
}

type JobResult struct {
	JobID    string
	Status   JobStatus
	Error    string
	Result   interface{}
	Progress float64
}

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
	}

	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,
	}
	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)

	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
	}

	return map[string]interface{}{
		"message":    "scan completed",
		"library_id": libraryID,
	}, 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()
}