package services import ( "bookhoard/internal/database" "context" "fmt" "sync" "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 } 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.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 := NewEbookScanner(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 { 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.cancel() close(w.jobQueue) w.wg.Wait() }