yonlu/yellowjacket
1
1
Fork 0
Code Issues 55 Pull Requests 6 Actions Packages Projects Releases 12 Wiki Activity

feat(11-01): per-library scan pipeline with queue coordinator

Browse Source 
Task 1: Schema, events, and progress types
- Add library_id to CreateAudioFile SQL INSERT and regenerate sqlc code
- Add LibraryScanQueued and LibraryScanQueueDrained event constants
- Regenerate TypeScript events via genevents
- Add LibraryID, LibraryName, QueuedCount to ScanProgress
- Add LibraryID, LibraryName to ScanMetrics
- Add libraryID field to importResult for threading through pipeline

Task 2: Scan queue coordinator and per-library scanning
- Create scan_queue.go with ScanLibrary(id), ScanAllLibraries()
- Add CancelCurrentScan(), CancelAllScans() for queue-aware cancellation
- FIFO scan queue with silent dedup (same library already scanning or queued)
- Refactor Scan() -> scanInternal(libraryID, libraryName, libraryPath)
- Replace GetAllAudioFiles with GetAudioFilesByLibrary for per-library loading
- Thread libraryID through DB writer to set CreateAudioFileParams.LibraryID
- drainQueue auto-starts next queued library or emits LibraryScanQueueDrained
- Pause freezes current scan AND queue
- Add GetScanQueueLength() and QueuedLibraryNames() for UI
- Mark CancelScan() and Scan() as deprecated
This commit is contained in:
yonlu
2026-03-09 16:02:54 -04:00
parent 68463f3549
commit 943db1cf27
11 changed files with 357 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/database/sql/queries/audio_files.sql
+1 -1
View File
@@ -1,5 +1,5 @@
-- name: CreateAudioFile :one
INSERT INTO audio_files (file_path, length_milliseconds, file_type_id, recording_id, sample_rate, bit_depth, channels, bitrate, file_size, basename) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
INSERT INTO audio_files (file_path, length_milliseconds, file_type_id, recording_id, sample_rate, bit_depth, channels, bitrate, file_size, basename, library_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
RETURNING *;
-- name: GetAudioFile :one
backend/database/sql/sqlcgen/audio_files.sql.go
+3 -1
View File
@@ -34,7 +34,7 @@ func (q *Queries) CountAudioFilesByLibrary(ctx context.Context, libraryID int64)
}
const createAudioFile = `-- name: CreateAudioFile :one
INSERT INTO audio_files (file_path, length_milliseconds, file_type_id, recording_id, sample_rate, bit_depth, channels, bitrate, file_size, basename) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
INSERT INTO audio_files (file_path, length_milliseconds, file_type_id, recording_id, sample_rate, bit_depth, channels, bitrate, file_size, basename, library_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
RETURNING id, file_path, length_milliseconds, file_type_id, recording_id, sample_rate, bit_depth, channels, bitrate, file_size, basename, library_id
`
@@ -49,6 +49,7 @@ type CreateAudioFileParams struct {
Bitrate int64
FileSize int64
Basename string
LibraryID int64
}
func (q *Queries) CreateAudioFile(ctx context.Context, arg CreateAudioFileParams) (AudioFile, error) {
@@ -63,6 +64,7 @@ func (q *Queries) CreateAudioFile(ctx context.Context, arg CreateAudioFileParams
arg.Bitrate,
arg.FileSize,
arg.Basename,
arg.LibraryID,
)
var i AudioFile
err := row.Scan(
backend/events/events.go
+6
View File
@@ -54,3 +54,9 @@ const (
LibraryScanPaused = "LibraryScanPaused"
LibraryScanResumed = "LibraryScanResumed"
)
// Scan queue events.
const (
LibraryScanQueued = "LibraryScanQueued"
LibraryScanQueueDrained = "LibraryScanQueueDrained"
)
backend/library/library.go
+114 -51
View File
@@ -90,6 +90,11 @@ type Library struct {
scanCancel context.CancelFunc
scanPaused bool
scanPauseCh chan struct{}
// Scan queue fields — protected by mu.
scanQueue []scanQueueEntry
currentScanLibraryID int64
currentScanLibraryName string
}
// SetRescanHooks provides optional hooks for cross-cutting
@@ -174,12 +179,40 @@ func (l *Library) registerEventHandlers() {
})
}
// Scan syncs the library by adding new files and removing deleted ones.
// Files that exist but have incomplete metadata (recording_id = 0)
// will be updated. The returned ScanMetrics contains timing and
// count data for every phase of the scan.
// Scan syncs the library using the legacy DirectoryPath config.
// Retained for backward compatibility with handleConfigUpdate.
//
// Deprecated: Use ScanLibrary(id) for per-library scanning.
func (l *Library) Scan() (*ScanMetrics, error) {
if len(l.conf.DirectoryPath) == 0 {
return newScanMetrics(), errLibraryDirNotConfigured
}
lib, err := l.db.Queries.GetLibraryByPath(
l.ctx, string(l.conf.DirectoryPath),
)
if err != nil {
return newScanMetrics(), fmt.Errorf(
"could not resolve library for path %s: %w",
l.conf.DirectoryPath, err,
)
}
return l.scanInternal(lib.ID, lib.Name, lib.Path), nil
}
// scanInternal performs the full scan pipeline for a single library.
// It is called from the scan queue coordinator (startScan) or the
// legacy Scan() wrapper. The caller is responsible for goroutine
// management; this method blocks until the scan completes.
func (l *Library) scanInternal(
libraryID int64,
libraryName string,
libraryPath string,
) *ScanMetrics {
metrics := newScanMetrics()
metrics.LibraryID = libraryID
metrics.LibraryName = libraryName
scanStart := time.Now()
scanCtx, scanCancel := context.WithCancel(l.ctx)
@@ -195,7 +228,6 @@ func (l *Library) Scan() (*ScanMetrics, error) {
defer func() {
l.mu.Lock()
l.scanCancel = nil
l.scanActive = false
// If still paused, unpause so no dangling channel.
if l.scanPaused {
l.scanPaused = false
@@ -208,28 +240,54 @@ func (l *Library) Scan() (*ScanMetrics, error) {
l.mu.Unlock()
}()
if len(l.conf.DirectoryPath) == 0 {
return metrics, errLibraryDirNotConfigured
}
workerCount := resolveScanWorkerCount(
l.conf.ScanConcurrency,
string(l.conf.DirectoryPath),
ScanConcurrencyAuto,
libraryPath,
)
l.logger.Info(
"beginning library scan",
"libraryID", libraryID,
"libraryName", libraryName,
"libraryPath", libraryPath,
"workers", workerCount,
"concurrencyMode", l.conf.ScanConcurrency,
)
runtime.EventsEmit(l.ctx, events.LibraryScanStarted)
// Helper to build a ScanProgress with library identification.
queuedCount := func() int {
l.mu.Lock()
defer l.mu.Unlock()
basePath := string(l.conf.DirectoryPath)
return len(l.scanQueue)
}
mkProgress := func(
phase string,
total, processed, a, s, u int64,
) ScanProgress {
return ScanProgress{
Phase: phase,
Total: total,
Processed: processed,
Added: a,
Skipped: s,
Updated: u,
LibraryID: libraryID,
LibraryName: libraryName,
QueuedCount: queuedCount(),
}
}
runtime.EventsEmit(l.ctx, events.LibraryScanStarted, map[string]any{
"libraryId": libraryID,
"libraryName": libraryName,
})
basePath := libraryPath
// --- Pre-walk: count audio files for progress reporting ---
runtime.EventsEmit(l.ctx, events.LibraryScanProgress,
ScanProgress{Phase: "counting"},
mkProgress("counting", 0, 0, 0, 0, 0),
)
totalFiles := countAudioFiles(basePath)
@@ -239,14 +297,20 @@ func (l *Library) Scan() (*ScanMetrics, error) {
"total", totalFiles,
)
// --- Phase 1: load existing files from DB ---
// --- Phase 1: load existing files from DB (per-library) ---
loadStart := time.Now()
existingFiles, err := l.db.Queries.GetAllAudioFiles(l.ctx)
existingFiles, err := l.db.Queries.GetAudioFilesByLibrary(
l.ctx, libraryID,
)
if err != nil {
return metrics, fmt.Errorf(
"could not load existing audio files: %w", err,
l.logger.Error(
"could not load existing audio files",
"libraryID", libraryID,
"err", err,
)
return metrics
}
existingPaths := &sync.Map{}
@@ -259,7 +323,8 @@ func (l *Library) Scan() (*ScanMetrics, error) {
l.logger.Debug(
"loaded existing files from database",
"count", len(existingFiles),
"library-directory", l.conf.DirectoryPath,
"libraryID", libraryID,
"libraryPath", libraryPath,
)
workChan := make(chan scanWork, 100)
@@ -424,14 +489,10 @@ func (l *Library) Scan() (*ScanMetrics, error) {
runtime.EventsEmit(
l.ctx,
events.LibraryScanProgress,
ScanProgress{
Phase: "scanning",
Total: totalFiles,
Processed: a + s + u,
Added: a,
Skipped: s,
Updated: u,
},
mkProgress(
"scanning", totalFiles,
a+s+u, a, s, u,
),
)
case <-stopProgress:
return
@@ -477,6 +538,9 @@ func (l *Library) Scan() (*ScanMetrics, error) {
}
for result := range resultChan {
// Thread library ID into each result for saveAudioFile.
result.libraryID = libraryID
if !dbStarted {
dbStartVal = time.Now()
dbStarted = true
@@ -553,14 +617,7 @@ func (l *Library) Scan() (*ScanMetrics, error) {
u := updated.Load()
runtime.EventsEmit(l.ctx, events.LibraryScanProgress,
ScanProgress{
Phase: "scanning",
Total: totalFiles,
Processed: a + s + u,
Added: a,
Skipped: s,
Updated: u,
},
mkProgress("scanning", totalFiles, a+s+u, a, s, u),
)
// Close thumbnail channel and wait for all thumbnail workers
@@ -568,14 +625,9 @@ func (l *Library) Scan() (*ScanMetrics, error) {
// point so it is safe to close.
thumbStart := time.Now()
runtime.EventsEmit(l.ctx, events.LibraryScanProgress, ScanProgress{
Phase: "thumbnails",
Total: totalFiles,
Processed: a + s + u,
Added: a,
Skipped: s,
Updated: u,
})
runtime.EventsEmit(l.ctx, events.LibraryScanProgress,
mkProgress("thumbnails", totalFiles, a+s+u, a, s, u),
)
close(thumbChan)
thumbWg.Wait()
@@ -594,10 +646,9 @@ func (l *Library) Scan() (*ScanMetrics, error) {
l.logger.Info("scan cancelled, skipping orphan cleanup")
} else {
// --- Phase 5: orphan cleanup ---
runtime.EventsEmit(l.ctx, events.LibraryScanProgress, ScanProgress{
Phase: "orphans", Total: totalFiles,
Processed: a + s + u, Added: a, Skipped: s, Updated: u,
})
runtime.EventsEmit(l.ctx, events.LibraryScanProgress,
mkProgress("orphans", totalFiles, a+s+u, a, s, u),
)
orphanStart := time.Now()
@@ -669,26 +720,36 @@ func (l *Library) Scan() (*ScanMetrics, error) {
metrics.Removed = removed.Load()
metrics.Total = time.Since(scanStart)
if scanErr != nil {
l.logger.Warn(
"scan completed with errors",
"err", scanErr,
)
}
l.logger.Info(
"library scan complete",
"libraryID", libraryID,
"libraryName", libraryName,
"added", metrics.Added,
"updated", metrics.Updated,
"removed", metrics.Removed,
"skipped", metrics.Skipped,
"cancelled", cancelled,
"total", metrics.Total,
"library", l.conf.DirectoryPath,
)
if cancelled {
runtime.EventsEmit(l.ctx, events.LibraryScanCancelled, metrics)
runtime.EventsEmit(
l.ctx, events.LibraryScanCancelled, metrics,
)
} else {
runtime.EventsEmit(
l.ctx, events.LibraryScanComplete, metrics,
)
}
return metrics, scanErr
return metrics
}
// progressInterval controls how often scan progress events are
@@ -765,6 +826,7 @@ type importResult struct {
audioProps *metadata.AudioProperties
existingFileID int64 // non-zero if this is an update
needsUpdate bool
libraryID int64 // library this file belongs to
}
// extractAudioMetadata reads and extracts metadata from an audio file.
@@ -939,6 +1001,7 @@ func (l *Library) saveAudioFile(
Bitrate: int64(props.Bitrate),
FileSize: props.FileSize,
Basename: basename,
LibraryID: result.libraryID,
})
if err != nil {
return fmt.Errorf(
backend/library/metrics.go
+13 -6
View File
@@ -53,6 +53,10 @@ type ScanMetrics struct {
// Cancelled is true when the scan was stopped via CancelScan.
Cancelled bool `json:"cancelled"`
// Library identification.
LibraryID int64 `json:"libraryId"` // library that was scanned
LibraryName string `json:"libraryName"` // display name of scanned library
// Non-fatal issues encountered during scanning.
Warnings []ScanWarning `json:"warnings"`
}
@@ -60,12 +64,15 @@ type ScanMetrics struct {
// ScanProgress is the payload emitted periodically during a scan to
// report live progress to the frontend.
type ScanProgress struct {
Phase string `json:"phase"` // "counting", "scanning", "orphans", "thumbnails"
Total int64 `json:"total"` // total audio files from pre-walk count
Processed int64 `json:"processed"` // added + skipped + updated so far
Added int64 `json:"added"`
Skipped int64 `json:"skipped"`
Updated int64 `json:"updated"`
Phase string `json:"phase"` // "counting", "scanning", "orphans", "thumbnails"
Total int64 `json:"total"` // total audio files from pre-walk count
Processed int64 `json:"processed"` // added + skipped + updated so far
Added int64 `json:"added"`
Skipped int64 `json:"skipped"`
Updated int64 `json:"updated"`
LibraryID int64 `json:"libraryId"` // library being scanned
LibraryName string `json:"libraryName"` // display name of library being scanned
QueuedCount int `json:"queuedCount"` // number of libraries still queued after this one
}
// ScanWarning represents a non-fatal issue encountered during scanning.
backend/library/scan_control.go
+3
View File
@@ -10,6 +10,9 @@ import (
// CancelScan cancels an in-progress scan. Returns immediately;
// scan goroutines stop at their next checkpoint.
//
// Deprecated: Use CancelCurrentScan or CancelAllScans for
// queue-aware cancellation.
func (l *Library) CancelScan() {
l.mu.Lock()
cancel := l.scanCancel
backend/library/scan_queue.go
+173
View File
@@ -0,0 +1,173 @@
package library
import (
"fmt"
"github.com/wailsapp/wails/v2/pkg/runtime"
"yellowjacket/backend/events"
)
// scanQueueEntry holds the metadata needed to scan a single library.
type scanQueueEntry struct {
libraryID int64
libraryName string
libraryPath string
}
// ScanLibrary queues a scan for the library with the given database ID.
// If no scan is active the library is scanned immediately; otherwise it
// is appended to the queue. Duplicate requests (same library already
// scanning or already queued) are silently ignored.
func (l *Library) ScanLibrary(id int64) error {
lib, err := l.db.Queries.GetLibrary(l.ctx, id)
if err != nil {
return fmt.Errorf("could not get library %d: %w", id, err)
}
l.mu.Lock()
defer l.mu.Unlock()
// Silent dedup: already scanning this library.
if l.currentScanLibraryID == id {
return nil
}
// Silent dedup: already queued.
for _, entry := range l.scanQueue {
if entry.libraryID == id {
return nil
}
}
entry := scanQueueEntry{
libraryID: lib.ID,
libraryName: lib.Name,
libraryPath: lib.Path,
}
if !l.scanActive {
l.scanActive = true
l.currentScanLibraryID = entry.libraryID
l.currentScanLibraryName = entry.libraryName
go l.startScan(entry)
return nil
}
// A scan is already running — queue this library.
l.scanQueue = append(l.scanQueue, entry)
runtime.EventsEmit(l.ctx, events.LibraryScanQueued, map[string]any{
"libraryId": lib.ID,
"libraryName": lib.Name,
"queueLength": len(l.scanQueue),
})
return nil
}
// ScanAllLibraries queries all libraries from the database and queues
// each one for scanning. Existing dedup logic ensures no duplicates.
func (l *Library) ScanAllLibraries() error {
libs, err := l.db.Queries.GetAllLibraries(l.ctx)
if err != nil {
return fmt.Errorf("could not get all libraries: %w", err)
}
for _, lib := range libs {
if err := l.ScanLibrary(lib.ID); err != nil {
l.logger.Warn(
"could not queue library for scan",
"libraryID", lib.ID,
"libraryName", lib.Name,
"err", err,
)
}
}
return nil
}
// CancelCurrentScan cancels only the currently scanning library.
// The next queued library (if any) starts automatically when the
// current scan's goroutine completes.
func (l *Library) CancelCurrentScan() {
l.mu.Lock()
cancel := l.scanCancel
l.mu.Unlock()
if cancel != nil {
cancel()
}
}
// CancelAllScans cancels the current scan and clears the entire
// queue so no further libraries are scanned.
func (l *Library) CancelAllScans() {
l.mu.Lock()
l.scanQueue = nil
cancel := l.scanCancel
l.mu.Unlock()
if cancel != nil {
cancel()
}
}
// GetScanQueueLength returns the number of libraries waiting in the
// scan queue (excludes the currently scanning library).
func (l *Library) GetScanQueueLength() int {
l.mu.Lock()
defer l.mu.Unlock()
return len(l.scanQueue)
}
// QueuedLibraryNames returns the display names of libraries waiting
// in the scan queue, in FIFO order.
func (l *Library) QueuedLibraryNames() []string {
l.mu.Lock()
defer l.mu.Unlock()
names := make([]string, len(l.scanQueue))
for i, entry := range l.scanQueue {
names[i] = entry.libraryName
}
return names
}
// startScan runs the scan for a single library entry and then drains
// the queue. It is always called in a new goroutine.
func (l *Library) startScan(entry scanQueueEntry) {
l.scanInternal(entry.libraryID, entry.libraryName, entry.libraryPath)
l.drainQueue()
}
// drainQueue is called after each scan completes. If the queue is
// non-empty the next entry is popped and scanned; otherwise the
// scan pipeline is marked idle.
func (l *Library) drainQueue() {
l.mu.Lock()
if len(l.scanQueue) > 0 {
next := l.scanQueue[0]
l.scanQueue = l.scanQueue[1:]
l.currentScanLibraryID = next.libraryID
l.currentScanLibraryName = next.libraryName
l.mu.Unlock()
go l.startScan(next)
return
}
l.currentScanLibraryID = 0
l.currentScanLibraryName = ""
l.scanActive = false
l.mu.Unlock()
runtime.EventsEmit(l.ctx, events.LibraryScanQueueDrained)
}
frontend/src/events.ts
+4
View File
@@ -38,6 +38,10 @@ export const Events = {
LibraryScanCancelled: "LibraryScanCancelled",
LibraryScanPaused: "LibraryScanPaused",
LibraryScanResumed: "LibraryScanResumed",
// Scan queue events
LibraryScanQueued: "LibraryScanQueued",
LibraryScanQueueDrained: "LibraryScanQueueDrained",
} as const;
export type EventName = (typeof Events)[keyof typeof Events];
frontend/wailsjs/go/library/Library.d.ts
Vendored
+12
View File
@@ -3,6 +3,10 @@
import {library} from '../models';
import {context} from '../models';
export function CancelAllScans():Promise<void>;
export function CancelCurrentScan():Promise<void>;
export function CancelScan():Promise<void>;
export function FullRescan():Promise<library.ScanMetrics>;
@@ -19,6 +23,8 @@ export function GetAllGenresWithCounts():Promise<Array<library.GenreWithCount>>;
export function GetAllTracks():Promise<Array<library.Track>>;
export function GetScanQueueLength():Promise<number>;
export function GetTracksByGenre(arg1:string):Promise<Array<library.Track>>;
export function IsScanActive():Promise<boolean>;
@@ -27,10 +33,16 @@ export function IsScanPaused():Promise<boolean>;
export function PauseScan():Promise<void>;
export function QueuedLibraryNames():Promise<Array<string>>;
export function ResumeScan():Promise<void>;
export function Scan():Promise<library.ScanMetrics>;
export function ScanAllLibraries():Promise<void>;
export function ScanLibrary(arg1:number):Promise<void>;
export function SearchTracks(arg1:string):Promise<Array<library.Track>>;
export function SetContext(arg1:context.Context):Promise<void>;
frontend/wailsjs/go/library/Library.js
+24
View File
@@ -2,6 +2,14 @@
// Cynhyrchwyd y ffeil hon yn awtomatig. PEIDIWCH Â MODIWL
// This file is automatically generated. DO NOT EDIT
export function CancelAllScans() {
return window['go']['library']['Library']['CancelAllScans']();
}
export function CancelCurrentScan() {
return window['go']['library']['Library']['CancelCurrentScan']();
}
export function CancelScan() {
return window['go']['library']['Library']['CancelScan']();
}
@@ -34,6 +42,10 @@ export function GetAllTracks() {
return window['go']['library']['Library']['GetAllTracks']();
}
export function GetScanQueueLength() {
return window['go']['library']['Library']['GetScanQueueLength']();
}
export function GetTracksByGenre(arg1) {
return window['go']['library']['Library']['GetTracksByGenre'](arg1);
}
@@ -50,6 +62,10 @@ export function PauseScan() {
return window['go']['library']['Library']['PauseScan']();
}
export function QueuedLibraryNames() {
return window['go']['library']['Library']['QueuedLibraryNames']();
}
export function ResumeScan() {
return window['go']['library']['Library']['ResumeScan']();
}
@@ -58,6 +74,14 @@ export function Scan() {
return window['go']['library']['Library']['Scan']();
}
export function ScanAllLibraries() {
return window['go']['library']['Library']['ScanAllLibraries']();
}
export function ScanLibrary(arg1) {
return window['go']['library']['Library']['ScanLibrary'](arg1);
}
export function SearchTracks(arg1) {
return window['go']['library']['Library']['SearchTracks'](arg1);
}
frontend/wailsjs/go/models.ts
+4
View File
@@ -109,6 +109,8 @@ export namespace library {
skipped: number;
removed: number;
cancelled: boolean;
libraryId: number;
libraryName: string;
warnings: ScanWarning[];
static createFrom(source: any = {}) {
@@ -143,6 +145,8 @@ export namespace library {
this.skipped = source["skipped"];
this.removed = source["removed"];
this.cancelled = source["cancelled"];
this.libraryId = source["libraryId"];
this.libraryName = source["libraryName"];
this.warnings = this.convertValues(source["warnings"], ScanWarning);
}