// Package jobs provides a central registry for long-running background // work — library scans, search index builds, and anything else that runs // while the user is doing something else. Producers report progress // through a Handle; the registry coalesces those updates into a single // JobsChanged event so the frontend can render one indicator, one job // list, and one log viewer regardless of which subsystem is working. package jobs import ( "context" "log/slog" "sync" "sync/atomic" "time" "yellowjacket/backend/events" ) // Kind identifies the subsystem that owns a job. The frontend uses it // to pick an icon and to route "view details" to the right panel. type Kind string // Job kinds. const ( KindLibraryScan Kind = "library-scan" KindIndexBuild Kind = "index-build" KindDownload Kind = "download" KindAutotagApply Kind = "autotag-apply" // KindCatalogEnrich is background catalog work for content the user // already owns — the discography backfills. It is distinct from // KindIndexBuild because the two differ in what cancelling costs: // an index build discards hours of downloading and the frontend // confirms before stopping one, where a backfill is resumable per // artist and stopping it is free. KindCatalogEnrich Kind = "catalog-enrich" ) // State is the lifecycle position of a job. type State string // Job states. Queued, Running, Paused and Pausing are live; Complete, // Cancelled and Error are terminal. const ( StateQueued State = "queued" StateRunning State = "running" StatePausing State = "pausing" StatePaused State = "paused" StateCancelling State = "cancelling" StateComplete State = "complete" StateCancelled State = "cancelled" StateError State = "error" ) // IsTerminal reports whether the state means the job will not progress // further without being started again from scratch. func (s State) IsTerminal() bool { return s == StateComplete || s == StateCancelled || s == StateError } // Level is the severity of a job log entry. type Level string // Log levels. const ( LevelInfo Level = "info" LevelWarn Level = "warn" LevelError Level = "error" ) // maxLogEntries bounds the per-job log ring buffer. Scans can emit a // warning per unreadable file, so the buffer is a tail, not an archive. const maxLogEntries = 500 // emitInterval is how often a dirty registry is flushed to the frontend. // Progress tickers run at 300ms, so this keeps re-render cost bounded // without making the UI feel laggy. const emitInterval = 250 * time.Millisecond // finishedRetention is how long terminal jobs stay in the registry so // the user can read their logs after the fact. const finishedRetention = 30 * time.Minute // maxFinished caps how many terminal jobs are retained regardless of age. const maxFinished = 25 // Caps describes which controls a job supports. The frontend renders // buttons from these rather than switching on Kind, so a job that gains // pause support later needs no frontend change. type Caps struct { Pausable bool `json:"pausable"` Cancellable bool `json:"cancellable"` } // Stage is one named sub-step of a multi-stage job, such as an index // build tier. Jobs with a single linear phase leave Stages empty. type Stage struct { Name string `json:"name"` State string `json:"state"` // pending, running, complete, error, skipped Current int64 `json:"current"` Total int64 `json:"total"` Error string `json:"error,omitempty"` } // Stat is a display-only key/value pair shown in the job detail panel // (e.g. "Added" / "1,204"). type Stat struct { Label string `json:"label"` Value string `json:"value"` } // LogEntry is one line of a job's output log. type LogEntry struct { Time int64 `json:"time"` // unix milliseconds Level Level `json:"level"` Message string `json:"message"` Detail string `json:"detail,omitempty"` } // Job is the frontend-facing snapshot of a single background job. type Job struct { ID string `json:"id"` Kind Kind `json:"kind"` Title string `json:"title"` Subtitle string `json:"subtitle,omitempty"` State State `json:"state"` Phase string `json:"phase,omitempty"` // Current/Total drive the progress bar. Total == 0 means the job // is indeterminate and the frontend should render a spinner. Current int64 `json:"current"` Total int64 `json:"total"` Caps Caps `json:"caps"` Stages []Stage `json:"stages"` Stats []Stat `json:"stats"` Error string `json:"error,omitempty"` StartedAt int64 `json:"startedAt"` // unix milliseconds UpdatedAt int64 `json:"updatedAt"` EndedAt int64 `json:"endedAt,omitempty"` LogCount int `json:"logCount"` WarnCount int `json:"warnCount"` ErrorCount int `json:"errorCount"` } // Controls holds the callbacks the registry invokes when the user asks // for a job to be paused, resumed, or cancelled. All three are optional; // a nil callback means the corresponding capability is unavailable. // // Callbacks are invoked on a dedicated goroutine, so implementations may // block (StopBuild waits for its build goroutine to exit, for instance) // without stalling the Wails call dispatcher. type Controls struct { Pause func() Resume func() Cancel func() } // Spec describes a job at registration time. type Spec struct { ID string Kind Kind Title string Subtitle string Total int64 State State Caps Caps Controls Controls // Durable marks a job whose paused state should survive an app // restart. On the next launch the owning subsystem adopts it back // into the registry as paused instead of silently resuming. Durable bool } // Handle is a producer's write side of a registered job. Every mutator // marks the registry dirty; the emitter coalesces those into one event. type Handle struct { reg *Registry id string mu sync.Mutex job Job controls Controls durable bool log []LogEntry } // Registry owns every known job and pushes coalesced snapshots to the // frontend. It is safe for concurrent use. type Registry struct { logger *slog.Logger store *Store mu sync.RWMutex ctx context.Context jobs map[string]*Handle order []string dirty atomic.Bool } // NewRegistry creates a registry. Pass a nil store to disable // pause-across-restart persistence (tests do this). func NewRegistry(logger *slog.Logger, store *Store) *Registry { return &Registry{ logger: logger, store: store, jobs: make(map[string]*Handle), } } // SetContext injects the Wails runtime context and starts the coalescing // emitter. Until it is called, updates are recorded but not pushed. func (r *Registry) SetContext(ctx context.Context) { r.mu.Lock() r.ctx = ctx r.mu.Unlock() go r.emitLoop(ctx) } // emitLoop flushes the registry to the frontend whenever it is dirty. func (r *Registry) emitLoop(ctx context.Context) { ticker := time.NewTicker(emitInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: if r.dirty.Swap(false) { r.emit() } } } } // emit pushes a full snapshot to the frontend. A full snapshot (rather // than a delta) means a component that mounts mid-scan is correct from // the first event it receives. func (r *Registry) emit() { r.mu.RLock() ctx := r.ctx r.mu.RUnlock() if ctx == nil { return } events.Emit(ctx, events.JobsChanged, r.Snapshot()) } // touch marks the registry dirty so the next emitter tick publishes it. func (r *Registry) touch() { r.dirty.Store(true) } // flush publishes immediately. Used for state transitions, where a // quarter-second of lag would make a button press feel unresponsive. func (r *Registry) flush() { r.dirty.Store(false) r.emit() } // Start registers a job and returns its handle. Re-registering an // existing ID reuses the handle and its log, which is what happens when // a queued scan is popped off the queue and actually begins. func (r *Registry) Start(spec Spec) *Handle { now := nowMillis() state := spec.State if state == "" { state = StateRunning } r.mu.Lock() h, existing := r.jobs[spec.ID] if !existing { h = &Handle{reg: r, id: spec.ID} r.jobs[spec.ID] = h r.order = append(r.order, spec.ID) } r.mu.Unlock() h.mu.Lock() if !existing { h.job = Job{ ID: spec.ID, StartedAt: now, Stages: []Stage{}, Stats: []Stat{}, } } h.job.Kind = spec.Kind h.job.Title = spec.Title h.job.Subtitle = spec.Subtitle h.job.State = state h.job.Caps = spec.Caps h.job.Total = spec.Total h.job.UpdatedAt = now h.job.EndedAt = 0 h.job.Error = "" h.controls = spec.Controls h.durable = spec.Durable h.mu.Unlock() r.persistPause(spec.ID, state) r.pruneFinished() r.flush() return h } // Get returns the handle for an ID, or nil when unknown. func (r *Registry) Get(id string) *Handle { r.mu.RLock() defer r.mu.RUnlock() return r.jobs[id] } // Snapshot returns every known job, oldest registration first. func (r *Registry) Snapshot() []Job { r.mu.RLock() out := make([]Job, 0, len(r.order)) handles := make([]*Handle, 0, len(r.order)) for _, id := range r.order { if h, ok := r.jobs[id]; ok { handles = append(handles, h) } } r.mu.RUnlock() for _, h := range handles { out = append(out, h.Snapshot()) } return out } // Logs returns the retained log tail for a job, oldest entry first. func (r *Registry) Logs(id string) []LogEntry { h := r.Get(id) if h == nil { return []LogEntry{} } h.mu.Lock() defer h.mu.Unlock() out := make([]LogEntry, len(h.log)) copy(out, h.log) return out } // HasActive reports whether any job is in a non-terminal state. func (r *Registry) HasActive() bool { for _, j := range r.Snapshot() { if !j.State.IsTerminal() { return true } } return false } // Pause asks the owning subsystem to pause a job. The job moves to // "pausing" immediately for UI feedback; the producer confirms the // transition to "paused" when it actually stops. func (r *Registry) Pause(id string) { h := r.Get(id) if h == nil { return } h.mu.Lock() pause := h.controls.Pause pausable := h.job.Caps.Pausable live := !h.job.State.IsTerminal() h.mu.Unlock() if pause == nil || !pausable || !live { return } h.SetState(StatePausing) h.Logf(LevelInfo, "Pause requested") go pause() } // Resume asks the owning subsystem to continue a paused job. func (r *Registry) Resume(id string) { h := r.Get(id) if h == nil { return } h.mu.Lock() resume := h.controls.Resume paused := h.job.State == StatePaused || h.job.State == StatePausing h.mu.Unlock() if resume == nil || !paused { return } h.Logf(LevelInfo, "Resume requested") go resume() } // Cancel asks the owning subsystem to abandon a job. func (r *Registry) Cancel(id string) { h := r.Get(id) if h == nil { return } h.mu.Lock() cancel := h.controls.Cancel cancellable := h.job.Caps.Cancellable live := !h.job.State.IsTerminal() h.mu.Unlock() if cancel == nil || !cancellable || !live { return } h.SetState(StateCancelling) h.Logf(LevelInfo, "Cancel requested") go cancel() } // Remove drops a job from the registry entirely, discarding its log. func (r *Registry) Remove(id string) { r.mu.Lock() delete(r.jobs, id) for i, existing := range r.order { if existing == id { r.order = append(r.order[:i], r.order[i+1:]...) break } } r.mu.Unlock() if r.store != nil { r.store.ClearPaused(id) } r.flush() } // ClearFinished drops every terminal job. Bound to the "clear" action // in the jobs panel. func (r *Registry) ClearFinished() { r.mu.Lock() kept := r.order[:0] for _, id := range r.order { h, ok := r.jobs[id] if !ok { continue } h.mu.Lock() terminal := h.job.State.IsTerminal() h.mu.Unlock() if terminal { delete(r.jobs, id) continue } kept = append(kept, id) } r.order = kept r.mu.Unlock() r.flush() } // pruneFinished evicts terminal jobs that are older than the retention // window, and trims the oldest when too many have accumulated. func (r *Registry) pruneFinished() { cutoff := nowMillis() - finishedRetention.Milliseconds() r.mu.Lock() defer r.mu.Unlock() var finished []string for _, id := range r.order { h, ok := r.jobs[id] if !ok { continue } h.mu.Lock() terminal := h.job.State.IsTerminal() ended := h.job.EndedAt h.mu.Unlock() if terminal && ended > 0 && ended < cutoff { delete(r.jobs, id) continue } if terminal { finished = append(finished, id) } } // Trim the oldest terminal jobs beyond the cap. if excess := len(finished) - maxFinished; excess > 0 { for _, id := range finished[:excess] { delete(r.jobs, id) } } kept := r.order[:0] for _, id := range r.order { if _, ok := r.jobs[id]; ok { kept = append(kept, id) } } r.order = kept } // persistPause writes or clears the durable pause record for a job so a // paused job comes back paused after a restart instead of silently // resuming (or silently never running again). func (r *Registry) persistPause(id string, state State) { if r.store == nil { return } h := r.Get(id) if h == nil { return } h.mu.Lock() durable := h.durable job := h.job h.mu.Unlock() if !durable { return } if state != StatePaused { r.store.ClearPaused(id) return } r.store.SetPaused(Persisted{ ID: job.ID, Kind: job.Kind, Title: job.Title, Subtitle: job.Subtitle, }) } // PausedEntries returns the jobs of the given kind that were paused when // the app last shut down. Subsystems call this during startup and adopt // each entry back into the registry with its controls attached. func (r *Registry) PausedEntries(kind Kind) []Persisted { if r.store == nil { return nil } return r.store.PausedEntries(kind) } // IsPersistentlyPaused reports whether the given job ID was paused when // the app last shut down. Subsystems check this before auto-starting // work at launch, so a paused job stays paused. func (r *Registry) IsPersistentlyPaused(id string) bool { if r.store == nil { return false } return r.store.IsPaused(id) } // --------------------------------------------------------------------------- // Handle // --------------------------------------------------------------------------- // Snapshot returns a copy of the job's current state. func (h *Handle) Snapshot() Job { h.mu.Lock() defer h.mu.Unlock() job := h.job job.Stages = make([]Stage, len(h.job.Stages)) copy(job.Stages, h.job.Stages) job.Stats = make([]Stat, len(h.job.Stats)) copy(job.Stats, h.job.Stats) return job } // State returns the job's current state. func (h *Handle) State() State { h.mu.Lock() defer h.mu.Unlock() return h.job.State } // SetState moves the job to a new state. Terminal states stamp EndedAt. // Transitions flush immediately so controls feel responsive. func (h *Handle) SetState(state State) { h.mu.Lock() if h.job.State == state { h.mu.Unlock() return } h.job.State = state h.job.UpdatedAt = nowMillis() if state.IsTerminal() { h.job.EndedAt = h.job.UpdatedAt } else { h.job.EndedAt = 0 } h.mu.Unlock() h.reg.persistPause(h.id, state) h.reg.flush() } // SetPhase records the human-readable phase label ("Scanning files"). func (h *Handle) SetPhase(phase string) { h.mu.Lock() changed := h.job.Phase != phase h.job.Phase = phase h.job.UpdatedAt = nowMillis() h.mu.Unlock() if changed { h.reg.flush() return } h.reg.touch() } // SetProgress updates the progress numerator and denominator. Pass a // total of zero to render the job as indeterminate. func (h *Handle) SetProgress(current, total int64) { h.mu.Lock() h.job.Current = current h.job.Total = total h.job.UpdatedAt = nowMillis() h.mu.Unlock() h.reg.touch() } // SetSubtitle updates the secondary line shown under the job title. func (h *Handle) SetSubtitle(subtitle string) { h.mu.Lock() h.job.Subtitle = subtitle h.job.UpdatedAt = nowMillis() h.mu.Unlock() h.reg.touch() } // SetStats replaces the job's display statistics. func (h *Handle) SetStats(stats []Stat) { h.mu.Lock() h.job.Stats = stats h.job.UpdatedAt = nowMillis() h.mu.Unlock() h.reg.touch() } // SetStages replaces the job's stage list. Used by multi-tier jobs such // as the index build. func (h *Handle) SetStages(stages []Stage) { h.mu.Lock() h.job.Stages = stages h.job.UpdatedAt = nowMillis() h.mu.Unlock() h.reg.touch() } // SetCaps updates which controls the job currently supports. func (h *Handle) SetCaps(caps Caps) { h.mu.Lock() h.job.Caps = caps h.mu.Unlock() h.reg.touch() } // Logf appends a line to the job's log ring buffer. func (h *Handle) Logf(level Level, message string) { h.logEntry(level, message, "") } // LogDetail appends a log line carrying a secondary detail string, such // as the file path a warning refers to. func (h *Handle) LogDetail(level Level, message, detail string) { h.logEntry(level, message, detail) } func (h *Handle) logEntry(level Level, message, detail string) { h.mu.Lock() if len(h.log) >= maxLogEntries { // Drop the oldest entry. Log volume is low enough (phase // transitions and per-file warnings) that the copy is cheaper // than maintaining an explicit ring index. h.log = append(h.log[:0], h.log[1:]...) } h.log = append(h.log, LogEntry{ Time: nowMillis(), Level: level, Message: message, Detail: detail, }) h.job.LogCount++ switch level { case LevelWarn: h.job.WarnCount++ case LevelError: h.job.ErrorCount++ case LevelInfo: } h.mu.Unlock() h.reg.touch() } // Complete marks the job finished successfully. func (h *Handle) Complete() { h.SetPhase("") h.SetState(StateComplete) } // Cancelled marks the job as abandoned at the user's request. func (h *Handle) Cancelled() { h.SetPhase("") h.SetState(StateCancelled) } // Fail marks the job as errored and records the message. func (h *Handle) Fail(err error) { h.mu.Lock() h.job.Error = err.Error() h.mu.Unlock() h.Logf(LevelError, err.Error()) h.SetState(StateError) } func nowMillis() int64 { return time.Now().UnixMilli() }