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)
This commit is contained in:
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user