From 07717f4f77d6782fb689cc12b8ad928b2d98642d Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Thu, 29 Jan 2026 09:49:48 -0500 Subject: [PATCH] feat(scanner): add background worker and scheduler for async scanning - Add Worker service with configurable worker pool for async job processing - Implement job queue with status tracking (pending, running, completed, failed, cancelled) - Add Scheduler service for auto-scanning based on user scan settings - Check scan settings every 5 minutes and schedule background scan jobs - Support multiple libraries with individual scan frequencies (15-1440 minutes) --- internal/services/scheduler.go | 210 +++++++++++++++++++++++++++++ internal/services/worker.go | 233 +++++++++++++++++++++++++++++++++ 2 files changed, 443 insertions(+) create mode 100644 internal/services/scheduler.go create mode 100644 internal/services/worker.go diff --git a/internal/services/scheduler.go b/internal/services/scheduler.go new file mode 100644 index 0000000..0013aa1 --- /dev/null +++ b/internal/services/scheduler.go @@ -0,0 +1,210 @@ +package services + +import ( + "bookmann/internal/database" + "context" + "fmt" + "log" + "sync" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +type Scheduler struct { + worker *Worker + db Database + timers map[string]*time.Timer + mu sync.RWMutex + scanSettings map[string]ScanSetting + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup +} + +type Database interface { + GetScanSettings(ctx context.Context, id pgtype.UUID) (database.GetScanSettingsRow, error) + ListLibraries(ctx context.Context) ([]database.ListLibrariesRow, error) + GetLibraryFolders(ctx context.Context, libraryID pgtype.UUID) ([]database.LibraryFolders, error) +} + +type ScanSettingRow struct { + ScanFrequencyMinutes pgtype.Int4 + AutoScanEnabled pgtype.Bool +} + +type Library struct { + ID pgtype.UUID + Name string +} + +type LibraryFolder struct { + ID pgtype.UUID + LibraryID pgtype.UUID + FolderPath string +} + +type ScanSetting struct { + UserID string + Enabled bool + Frequency int +} + +func NewScheduler(worker *Worker, db Database) *Scheduler { + ctx, cancel := context.WithCancel(context.Background()) + + return &Scheduler{ + worker: worker, + db: db, + timers: make(map[string]*time.Timer), + scanSettings: make(map[string]ScanSetting), + ctx: ctx, + cancel: cancel, + } +} + +func (s *Scheduler) Start() { + s.wg.Add(1) + go s.runSettingsChecker() +} + +func (s *Scheduler) Stop() { + s.cancel() + + s.mu.Lock() + for _, timer := range s.timers { + timer.Stop() + } + s.timers = make(map[string]*time.Timer) + s.mu.Unlock() + + s.wg.Wait() +} + +func (s *Scheduler) runSettingsChecker() { + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + s.checkAndScheduleScans() + case <-s.ctx.Done(): + return + } + } +} + +func (s *Scheduler) checkAndScheduleScans() { + ctx, cancel := context.WithTimeout(s.ctx, 30*time.Second) + defer cancel() + + libraries, err := s.db.ListLibraries(ctx) + if err != nil { + log.Printf("Error fetching libraries for scan scheduling: %v", err) + return + } + + for _, library := range libraries { + settings, err := s.db.GetScanSettings(ctx, library.ID) + if err != nil { + log.Printf("Error getting scan settings for library %s: %v", library.ID, err) + continue + } + + if !settings.AutoScanEnabled.Bool || settings.ScanFrequencyMinutes.Int32 < 15 { + continue + } + + userID := fmt.Sprintf("%x", library.ID.Bytes) + + s.mu.Lock() + currentSetting, exists := s.scanSettings[userID] + scanSetting := ScanSetting{ + UserID: userID, + Enabled: settings.AutoScanEnabled.Bool, + Frequency: int(settings.ScanFrequencyMinutes.Int32), + } + s.scanSettings[userID] = scanSetting + s.mu.Unlock() + + if !exists || currentSetting.Frequency != scanSetting.Frequency { + s.scheduleLibraryScan(ctx, library.ID, userID, scanSetting.Frequency) + } + } +} + +func (s *Scheduler) scheduleLibraryScan(ctx context.Context, libraryID pgtype.UUID, userID string, frequencyMinutes int) { + userIDStr := fmt.Sprintf("%x", libraryID.Bytes) + + s.mu.Lock() + defer s.mu.Unlock() + + timerID := userIDStr + "-scan" + + if existingTimer, exists := s.timers[timerID]; exists { + existingTimer.Stop() + } + + duration := time.Duration(frequencyMinutes) * time.Minute + timer := time.AfterFunc(duration, func() { + s.triggerScheduledScan(ctx, libraryID, userIDStr) + + s.scheduleLibraryScan(ctx, libraryID, userID, frequencyMinutes) + }) + + s.timers[timerID] = timer + + log.Printf("Scheduled scan for library %s every %d minutes", libraryID, frequencyMinutes) +} + +func (s *Scheduler) triggerScheduledScan(ctx context.Context, libraryID pgtype.UUID, userID string) { + log.Printf("Triggering scheduled scan for library %s", libraryID) + + folders, err := s.db.GetLibraryFolders(ctx, libraryID) + if err != nil { + log.Printf("Error getting folders for library %s: %v", libraryID, err) + return + } + + if len(folders) == 0 { + log.Printf("No folders configured for library %s, skipping scan", libraryID) + return + } + + folderPaths := make([]string, len(folders)) + for i, folder := range folders { + folderPaths[i] = folder.FolderPath + } + + job := &Job{ + ID: uuid.New().String(), + Type: JobTypeScan, + Params: map[string]interface{}{ + "library_id": fmt.Sprintf("%x", libraryID.Bytes), + "folders": folderPaths, + "admin_id": userID, + "db": s.db, + }, + Status: JobStatusPending, + Context: ctx, + } + + if err := s.worker.EnqueueJob(job); err != nil { + log.Printf("Error enqueuing scan job: %v", err) + } else { + log.Printf("Enqueued scan job %s for library %s", job.ID, libraryID) + } +} + +func (s *Scheduler) UpdateScanSettings(userID string, enabled bool, frequencyMinutes int) { + s.mu.Lock() + defer s.mu.Unlock() + + s.scanSettings[userID] = ScanSetting{ + UserID: userID, + Enabled: enabled, + Frequency: frequencyMinutes, + } +} diff --git a/internal/services/worker.go b/internal/services/worker.go new file mode 100644 index 0000000..2f14a3c --- /dev/null +++ b/internal/services/worker.go @@ -0,0 +1,233 @@ +package services + +import ( + "bookmann/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() +}