mirror of
https://github.com/navidrome/navidrome.git
synced 2026-10-27 21:57:17 +01:00
feat(library): add/remove watchers on demand, not only when server starts
Signed-off-by: Deluan <deluan@navidrome.org>
This commit is contained in:
parent
8c8ef7836a
commit
ee18a04da9
6 changed files with 361 additions and 57 deletions
|
|
@ -68,7 +68,8 @@ func CreateNativeAPIRouter(ctx context.Context) *nativeapi.Router {
|
|||
cacheWarmer := artwork.NewCacheWarmer(artworkArtwork, fileCache)
|
||||
broker := events.GetBroker()
|
||||
scannerScanner := scanner.New(ctx, dataStore, cacheWarmer, broker, playlists, metricsMetrics)
|
||||
library := core.NewLibrary(dataStore, scannerScanner)
|
||||
watcher := scanner.GetWatcher(dataStore, scannerScanner)
|
||||
library := core.NewLibrary(dataStore, scannerScanner, watcher)
|
||||
router := nativeapi.New(dataStore, share, playlists, insights, library)
|
||||
return router
|
||||
}
|
||||
|
|
@ -175,7 +176,7 @@ func CreateScanWatcher(ctx context.Context) scanner.Watcher {
|
|||
broker := events.GetBroker()
|
||||
playlists := core.NewPlaylists(dataStore)
|
||||
scannerScanner := scanner.New(ctx, dataStore, cacheWarmer, broker, playlists, metricsMetrics)
|
||||
watcher := scanner.NewWatcher(dataStore, scannerScanner)
|
||||
watcher := scanner.GetWatcher(dataStore, scannerScanner)
|
||||
return watcher
|
||||
}
|
||||
|
||||
|
|
@ -196,7 +197,7 @@ func getPluginManager() plugins.Manager {
|
|||
|
||||
// wire_injectors.go:
|
||||
|
||||
var allProviders = wire.NewSet(core.Set, artwork.Set, server.New, subsonic.New, nativeapi.New, public.New, persistence.New, lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, scanner.New, scanner.NewWatcher, plugins.GetManager, metrics.GetPrometheusInstance, db.Db, wire.Bind(new(agents.PluginLoader), new(plugins.Manager)), wire.Bind(new(scrobbler.PluginLoader), new(plugins.Manager)), wire.Bind(new(core.Scanner), new(scanner.Scanner)))
|
||||
var allProviders = wire.NewSet(core.Set, artwork.Set, server.New, subsonic.New, nativeapi.New, public.New, persistence.New, lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, scanner.New, scanner.GetWatcher, plugins.GetManager, metrics.GetPrometheusInstance, db.Db, wire.Bind(new(agents.PluginLoader), new(plugins.Manager)), wire.Bind(new(scrobbler.PluginLoader), new(plugins.Manager)), wire.Bind(new(core.Scanner), new(scanner.Scanner)), wire.Bind(new(core.Watcher), new(scanner.Watcher)))
|
||||
|
||||
func GetPluginManager(ctx context.Context) plugins.Manager {
|
||||
manager := getPluginManager()
|
||||
|
|
|
|||
|
|
@ -38,13 +38,14 @@ var allProviders = wire.NewSet(
|
|||
listenbrainz.NewRouter,
|
||||
events.GetBroker,
|
||||
scanner.New,
|
||||
scanner.NewWatcher,
|
||||
scanner.GetWatcher,
|
||||
plugins.GetManager,
|
||||
metrics.GetPrometheusInstance,
|
||||
db.Db,
|
||||
wire.Bind(new(agents.PluginLoader), new(plugins.Manager)),
|
||||
wire.Bind(new(scrobbler.PluginLoader), new(plugins.Manager)),
|
||||
wire.Bind(new(core.Scanner), new(scanner.Scanner)),
|
||||
wire.Bind(new(core.Watcher), new(scanner.Watcher)),
|
||||
)
|
||||
|
||||
func CreateDataStore() model.DataStore {
|
||||
|
|
|
|||
|
|
@ -19,11 +19,17 @@ import (
|
|||
"github.com/navidrome/navidrome/model/request"
|
||||
)
|
||||
|
||||
// Scanner defines the interface for triggering library scans
|
||||
// Scanner interface for triggering scans
|
||||
type Scanner interface {
|
||||
ScanAll(ctx context.Context, fullScan bool) (warnings []string, err error)
|
||||
}
|
||||
|
||||
// Watcher interface for managing file system watchers
|
||||
type Watcher interface {
|
||||
Watch(ctx context.Context, lib *model.Library) error
|
||||
StopWatching(ctx context.Context, libraryID int) error
|
||||
}
|
||||
|
||||
// Library provides business logic for library management and user-library associations
|
||||
type Library interface {
|
||||
GetUserLibraries(ctx context.Context, userID string) (model.Libraries, error)
|
||||
|
|
@ -36,11 +42,16 @@ type Library interface {
|
|||
type libraryService struct {
|
||||
ds model.DataStore
|
||||
scanner Scanner
|
||||
watcher Watcher
|
||||
}
|
||||
|
||||
// NewLibrary creates a new Library service
|
||||
func NewLibrary(ds model.DataStore, scanner Scanner) Library {
|
||||
return &libraryService{ds: ds, scanner: scanner}
|
||||
func NewLibrary(ds model.DataStore, scanner Scanner, watcher Watcher) Library {
|
||||
return &libraryService{
|
||||
ds: ds,
|
||||
scanner: scanner,
|
||||
watcher: watcher,
|
||||
}
|
||||
}
|
||||
|
||||
// User-library association operations
|
||||
|
|
@ -118,6 +129,7 @@ func (s *libraryService) NewRepository(ctx context.Context) rest.Repository {
|
|||
Repository: repo.(rest.Repository),
|
||||
ds: s.ds,
|
||||
scanner: s.scanner,
|
||||
watcher: s.watcher,
|
||||
}
|
||||
return wrapper
|
||||
}
|
||||
|
|
@ -128,6 +140,7 @@ type libraryRepositoryWrapper struct {
|
|||
ctx context.Context
|
||||
ds model.DataStore
|
||||
scanner Scanner
|
||||
watcher Watcher
|
||||
}
|
||||
|
||||
func (r *libraryRepositoryWrapper) Save(entity interface{}) (string, error) {
|
||||
|
|
@ -141,7 +154,13 @@ func (r *libraryRepositoryWrapper) Save(entity interface{}) (string, error) {
|
|||
return "", r.mapError(err)
|
||||
}
|
||||
|
||||
// Trigger scan after successful library creation
|
||||
// Start watcher and trigger scan after successful library creation
|
||||
if r.watcher != nil {
|
||||
if err := r.watcher.Watch(r.ctx, lib); err != nil {
|
||||
log.Warn(r.ctx, "Failed to start watcher for new library", "libraryID", lib.ID, "name", lib.Name, "path", lib.Path, err)
|
||||
}
|
||||
}
|
||||
|
||||
if r.scanner != nil {
|
||||
go r.triggerScan(lib, "new")
|
||||
}
|
||||
|
|
@ -174,9 +193,17 @@ func (r *libraryRepositoryWrapper) Update(id string, entity interface{}, cols ..
|
|||
return r.mapError(err)
|
||||
}
|
||||
|
||||
// Trigger scan if path was updated
|
||||
if pathChanged && r.scanner != nil {
|
||||
go r.triggerScan(lib, "updated")
|
||||
// Restart watcher and trigger scan if path was updated
|
||||
if pathChanged {
|
||||
if r.watcher != nil {
|
||||
if err := r.watcher.Watch(r.ctx, lib); err != nil {
|
||||
log.Warn(r.ctx, "Failed to restart watcher for updated library", "libraryID", lib.ID, "name", lib.Name, "path", lib.Path, err)
|
||||
}
|
||||
}
|
||||
|
||||
if r.scanner != nil {
|
||||
go r.triggerScan(lib, "updated")
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
@ -201,7 +228,13 @@ func (r *libraryRepositoryWrapper) Delete(id string) error {
|
|||
return r.mapError(err)
|
||||
}
|
||||
|
||||
// Trigger scan after successful library deletion to clean up orphaned data
|
||||
// Stop watcher and trigger scan after successful library deletion to clean up orphaned data
|
||||
if r.watcher != nil {
|
||||
if err := r.watcher.StopWatching(r.ctx, libID); err != nil {
|
||||
log.Warn(r.ctx, "Failed to stop watcher for deleted library", "libraryID", libID, "name", lib.Name, "path", lib.Path, err)
|
||||
}
|
||||
}
|
||||
|
||||
if r.scanner != nil {
|
||||
go r.triggerScan(lib, "deleted")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ var _ = Describe("Library Service", func() {
|
|||
var ctx context.Context
|
||||
var tempDir string
|
||||
var scanner *mockScanner
|
||||
var watcherManager *mockWatcherManager
|
||||
|
||||
BeforeEach(func() {
|
||||
DeferCleanup(configtest.SetupConfig())
|
||||
|
|
@ -40,7 +41,11 @@ var _ = Describe("Library Service", func() {
|
|||
|
||||
// Create a mock scanner that tracks calls
|
||||
scanner = &mockScanner{}
|
||||
service = core.NewLibrary(ds, scanner)
|
||||
// Create a mock watcher manager
|
||||
watcherManager = &mockWatcherManager{
|
||||
libraryStates: make(map[int]model.Library),
|
||||
}
|
||||
service = core.NewLibrary(ds, scanner, watcherManager)
|
||||
ctx = context.Background()
|
||||
|
||||
// Create a temporary directory for testing valid paths
|
||||
|
|
@ -714,6 +719,102 @@ var _ = Describe("Library Service", func() {
|
|||
return scanner.len()
|
||||
}, "100ms", "10ms").Should(Equal(0))
|
||||
})
|
||||
|
||||
Context("Watcher Integration", func() {
|
||||
It("starts watcher when creating a new library", func() {
|
||||
library := &model.Library{ID: 1, Name: "New Library", Path: tempDir}
|
||||
|
||||
_, err := repo.Save(library)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
|
||||
// Verify watcher was started
|
||||
Eventually(func() int {
|
||||
return watcherManager.lenStarted()
|
||||
}, "1s", "10ms").Should(Equal(1))
|
||||
|
||||
Expect(watcherManager.StartedWatchers[0].ID).To(Equal(1))
|
||||
Expect(watcherManager.StartedWatchers[0].Name).To(Equal("New Library"))
|
||||
Expect(watcherManager.StartedWatchers[0].Path).To(Equal(tempDir))
|
||||
})
|
||||
|
||||
It("restarts watcher when library path is updated", func() {
|
||||
// First create a library
|
||||
libraryRepo.SetData(model.Libraries{
|
||||
{ID: 1, Name: "Original Library", Path: tempDir},
|
||||
})
|
||||
|
||||
// Simulate that this library already has a watcher
|
||||
watcherManager.simulateExistingLibrary(model.Library{ID: 1, Name: "Original Library", Path: tempDir})
|
||||
|
||||
// Create a new temp directory for the update
|
||||
newTempDir, err := os.MkdirTemp("", "navidrome-library-update-")
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
DeferCleanup(func() { os.RemoveAll(newTempDir) })
|
||||
|
||||
// Update library with new path
|
||||
library := &model.Library{ID: 1, Name: "Updated Library", Path: newTempDir}
|
||||
err = repo.Update("1", library)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
|
||||
// Verify watcher was restarted
|
||||
Eventually(func() int {
|
||||
return watcherManager.lenRestarted()
|
||||
}, "1s", "10ms").Should(Equal(1))
|
||||
|
||||
Expect(watcherManager.RestartedWatchers[0].ID).To(Equal(1))
|
||||
Expect(watcherManager.RestartedWatchers[0].Path).To(Equal(newTempDir))
|
||||
})
|
||||
|
||||
It("does not restart watcher when only library name is updated", func() {
|
||||
// First create a library
|
||||
libraryRepo.SetData(model.Libraries{
|
||||
{ID: 1, Name: "Original Library", Path: tempDir},
|
||||
})
|
||||
|
||||
// Update library with same path but different name
|
||||
library := &model.Library{ID: 1, Name: "Updated Name", Path: tempDir}
|
||||
err := repo.Update("1", library)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
|
||||
// Verify watcher was NOT restarted (since path didn't change)
|
||||
Consistently(func() int {
|
||||
return watcherManager.lenRestarted()
|
||||
}, "100ms", "10ms").Should(Equal(0))
|
||||
})
|
||||
|
||||
It("stops watcher when library is deleted", func() {
|
||||
// Set up a library
|
||||
libraryRepo.SetData(model.Libraries{
|
||||
{ID: 1, Name: "Test Library", Path: tempDir},
|
||||
})
|
||||
|
||||
err := repo.Delete("1")
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
|
||||
// Verify watcher was stopped
|
||||
Eventually(func() int {
|
||||
return watcherManager.lenStopped()
|
||||
}, "1s", "10ms").Should(Equal(1))
|
||||
|
||||
Expect(watcherManager.StoppedWatchers[0]).To(Equal(1))
|
||||
})
|
||||
|
||||
It("does not stop watcher when library deletion fails", func() {
|
||||
// Set up a library
|
||||
libraryRepo.SetData(model.Libraries{
|
||||
{ID: 1, Name: "Test Library", Path: tempDir},
|
||||
})
|
||||
|
||||
// Mock deletion to fail by trying to delete non-existent library
|
||||
err := repo.Delete("999")
|
||||
Expect(err).To(HaveOccurred())
|
||||
|
||||
// Verify watcher was NOT stopped since deletion failed
|
||||
Consistently(func() int {
|
||||
return watcherManager.lenStopped()
|
||||
}, "100ms", "10ms").Should(Equal(0))
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
|
|
@ -741,3 +842,70 @@ func (m *mockScanner) len() int {
|
|||
defer m.mu.RUnlock()
|
||||
return len(m.ScanCalls)
|
||||
}
|
||||
|
||||
// mockWatcherManager provides a simple mock implementation of core.Watcher for testing
|
||||
type mockWatcherManager struct {
|
||||
StartedWatchers []model.Library
|
||||
StoppedWatchers []int
|
||||
RestartedWatchers []model.Library
|
||||
libraryStates map[int]model.Library // Track which libraries we know about
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
func (m *mockWatcherManager) Watch(ctx context.Context, lib *model.Library) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
// Check if we already know about this library ID
|
||||
if _, exists := m.libraryStates[lib.ID]; exists {
|
||||
// This is a restart - the library already existed
|
||||
// Update our tracking and record the restart
|
||||
for i, startedLib := range m.StartedWatchers {
|
||||
if startedLib.ID == lib.ID {
|
||||
m.StartedWatchers[i] = *lib
|
||||
break
|
||||
}
|
||||
}
|
||||
m.RestartedWatchers = append(m.RestartedWatchers, *lib)
|
||||
m.libraryStates[lib.ID] = *lib
|
||||
return nil
|
||||
}
|
||||
|
||||
// This is a new library - first time we're seeing it
|
||||
m.StartedWatchers = append(m.StartedWatchers, *lib)
|
||||
m.libraryStates[lib.ID] = *lib
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockWatcherManager) StopWatching(ctx context.Context, libraryID int) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.StoppedWatchers = append(m.StoppedWatchers, libraryID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockWatcherManager) lenStarted() int {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return len(m.StartedWatchers)
|
||||
}
|
||||
|
||||
func (m *mockWatcherManager) lenStopped() int {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return len(m.StoppedWatchers)
|
||||
}
|
||||
|
||||
func (m *mockWatcherManager) lenRestarted() int {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return len(m.RestartedWatchers)
|
||||
}
|
||||
|
||||
// simulateExistingLibrary simulates the scenario where a library already exists
|
||||
// and has a watcher running (used by tests to set up the initial state)
|
||||
func (m *mockWatcherManager) simulateExistingLibrary(lib model.Library) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.libraryStates[lib.ID] = lib
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,14 +28,14 @@ func NewMockLibraryService() *MockLibraryService {
|
|||
}
|
||||
}
|
||||
|
||||
func (m *MockLibraryService) GetAll(ctx context.Context) (model.Libraries, error) {
|
||||
func (m *MockLibraryService) getAll(ctx context.Context) (model.Libraries, error) {
|
||||
if m.Err != nil {
|
||||
return nil, m.Err
|
||||
}
|
||||
return m.Libraries, nil
|
||||
}
|
||||
|
||||
func (m *MockLibraryService) Get(ctx context.Context, id int) (*model.Library, error) {
|
||||
func (m *MockLibraryService) get(ctx context.Context, id int) (*model.Library, error) {
|
||||
if m.Err != nil {
|
||||
return nil, m.Err
|
||||
}
|
||||
|
|
@ -47,7 +47,7 @@ func (m *MockLibraryService) Get(ctx context.Context, id int) (*model.Library, e
|
|||
return nil, model.ErrNotFound
|
||||
}
|
||||
|
||||
func (m *MockLibraryService) Create(ctx context.Context, library *model.Library) error {
|
||||
func (m *MockLibraryService) create(ctx context.Context, library *model.Library) error {
|
||||
if m.Err != nil {
|
||||
return m.Err
|
||||
}
|
||||
|
|
@ -63,7 +63,7 @@ func (m *MockLibraryService) Create(ctx context.Context, library *model.Library)
|
|||
return nil
|
||||
}
|
||||
|
||||
func (m *MockLibraryService) Update(ctx context.Context, library *model.Library) error {
|
||||
func (m *MockLibraryService) update(ctx context.Context, id int, library *model.Library) error {
|
||||
if m.Err != nil {
|
||||
return m.Err
|
||||
}
|
||||
|
|
@ -76,7 +76,8 @@ func (m *MockLibraryService) Update(ctx context.Context, library *model.Library)
|
|||
|
||||
// Find and update in mock data
|
||||
for i, lib := range m.Libraries {
|
||||
if lib.ID == library.ID {
|
||||
if lib.ID == id {
|
||||
library.ID = id
|
||||
m.Libraries[i] = *library
|
||||
return nil
|
||||
}
|
||||
|
|
@ -84,7 +85,7 @@ func (m *MockLibraryService) Update(ctx context.Context, library *model.Library)
|
|||
return model.ErrNotFound
|
||||
}
|
||||
|
||||
func (m *MockLibraryService) Delete(ctx context.Context, id int) error {
|
||||
func (m *MockLibraryService) delete(ctx context.Context, id int) error {
|
||||
if m.Err != nil {
|
||||
return m.Err
|
||||
}
|
||||
|
|
@ -158,7 +159,7 @@ type mockLibraryRepository struct {
|
|||
}
|
||||
|
||||
func (r *mockLibraryRepository) Count(options ...rest.QueryOptions) (int64, error) {
|
||||
libs, err := r.service.GetAll(r.ctx)
|
||||
libs, err := r.service.getAll(r.ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
|
@ -170,7 +171,7 @@ func (r *mockLibraryRepository) Read(id string) (interface{}, error) {
|
|||
if err != nil {
|
||||
return nil, rest.ErrNotFound
|
||||
}
|
||||
lib, err := r.service.Get(r.ctx, idInt)
|
||||
lib, err := r.service.get(r.ctx, idInt)
|
||||
if errors.Is(err, model.ErrNotFound) {
|
||||
return nil, rest.ErrNotFound
|
||||
}
|
||||
|
|
@ -178,7 +179,7 @@ func (r *mockLibraryRepository) Read(id string) (interface{}, error) {
|
|||
}
|
||||
|
||||
func (r *mockLibraryRepository) ReadAll(options ...rest.QueryOptions) (interface{}, error) {
|
||||
return r.service.GetAll(r.ctx)
|
||||
return r.service.getAll(r.ctx)
|
||||
}
|
||||
|
||||
func (r *mockLibraryRepository) EntityName() string {
|
||||
|
|
@ -191,7 +192,7 @@ func (r *mockLibraryRepository) NewInstance() interface{} {
|
|||
|
||||
func (r *mockLibraryRepository) Save(entity interface{}) (string, error) {
|
||||
lib := entity.(*model.Library)
|
||||
err := r.service.Create(r.ctx, lib)
|
||||
err := r.service.create(r.ctx, lib)
|
||||
if errors.Is(err, model.ErrValidation) {
|
||||
return "", &rest.ValidationError{Errors: map[string]string{"validation": err.Error()}}
|
||||
}
|
||||
|
|
@ -202,13 +203,12 @@ func (r *mockLibraryRepository) Save(entity interface{}) (string, error) {
|
|||
}
|
||||
|
||||
func (r *mockLibraryRepository) Update(id string, entity interface{}, cols ...string) error {
|
||||
lib := entity.(*model.Library)
|
||||
idInt, err := strconv.Atoi(id)
|
||||
if err != nil {
|
||||
return &rest.ValidationError{Errors: map[string]string{"id": "invalid library ID format"}}
|
||||
}
|
||||
lib.ID = idInt
|
||||
err = r.service.Update(r.ctx, lib)
|
||||
lib := entity.(*model.Library)
|
||||
err = r.service.update(r.ctx, idInt, lib)
|
||||
if errors.Is(err, model.ErrNotFound) {
|
||||
return rest.ErrNotFound
|
||||
}
|
||||
|
|
@ -223,7 +223,7 @@ func (r *mockLibraryRepository) Delete(id string) error {
|
|||
if err != nil {
|
||||
return &rest.ValidationError{Errors: map[string]string{"id": "invalid library ID format"}}
|
||||
}
|
||||
err = r.service.Delete(r.ctx, idInt)
|
||||
err = r.service.delete(r.ctx, idInt)
|
||||
if errors.Is(err, model.ErrNotFound) {
|
||||
return rest.ErrNotFound
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,42 +5,67 @@ import (
|
|||
"fmt"
|
||||
"io/fs"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/core/storage"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/utils/singleton"
|
||||
)
|
||||
|
||||
type Watcher interface {
|
||||
Run(ctx context.Context) error
|
||||
Watch(ctx context.Context, lib *model.Library) error
|
||||
StopWatching(ctx context.Context, libraryID int) error
|
||||
}
|
||||
|
||||
type watcher struct {
|
||||
ds model.DataStore
|
||||
scanner Scanner
|
||||
triggerWait time.Duration
|
||||
mainCtx context.Context
|
||||
ds model.DataStore
|
||||
scanner Scanner
|
||||
triggerWait time.Duration
|
||||
watcherNotify chan model.Library
|
||||
libraryWatchers map[int]*libraryWatcherInstance
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
func NewWatcher(ds model.DataStore, s Scanner) Watcher {
|
||||
return &watcher{ds: ds, scanner: s, triggerWait: conf.Server.Scanner.WatcherWait}
|
||||
type libraryWatcherInstance struct {
|
||||
library *model.Library
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// GetWatcher returns the watcher singleton
|
||||
func GetWatcher(ds model.DataStore, s Scanner) Watcher {
|
||||
return singleton.GetInstance(func() *watcher {
|
||||
return &watcher{
|
||||
ds: ds,
|
||||
scanner: s,
|
||||
triggerWait: conf.Server.Scanner.WatcherWait,
|
||||
watcherNotify: make(chan model.Library, 1),
|
||||
libraryWatchers: make(map[int]*libraryWatcherInstance),
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (w *watcher) Run(ctx context.Context) error {
|
||||
// Keep the main context to be used in all watchers added later
|
||||
w.mainCtx = ctx
|
||||
|
||||
// Start watchers for all existing libraries
|
||||
libs, err := w.ds.Library(ctx).GetAll()
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting libraries: %w", err)
|
||||
}
|
||||
|
||||
watcherChan := make(chan struct{})
|
||||
defer close(watcherChan)
|
||||
|
||||
// Start a watcher for each library
|
||||
for _, lib := range libs {
|
||||
go watchLib(ctx, lib, watcherChan)
|
||||
if err := w.Watch(ctx, &lib); err != nil {
|
||||
log.Warn(ctx, "Failed to start watcher for existing library", "libraryID", lib.ID, "name", lib.Name, "path", lib.Path, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Main scan triggering loop
|
||||
trigger := time.NewTimer(w.triggerWait)
|
||||
trigger.Stop()
|
||||
waiting := false
|
||||
|
|
@ -68,61 +93,137 @@ func (w *watcher) Run(ctx context.Context) error {
|
|||
}
|
||||
}()
|
||||
case <-ctx.Done():
|
||||
// Stop all library watchers
|
||||
w.mu.Lock()
|
||||
for libraryID, instance := range w.libraryWatchers {
|
||||
log.Debug(ctx, "Stopping library watcher due to context cancellation", "libraryID", libraryID)
|
||||
instance.cancel()
|
||||
}
|
||||
w.libraryWatchers = make(map[int]*libraryWatcherInstance)
|
||||
w.mu.Unlock()
|
||||
return nil
|
||||
case <-watcherChan:
|
||||
case lib := <-w.watcherNotify:
|
||||
if !waiting {
|
||||
log.Debug(ctx, "Watcher: Detected changes. Waiting for more changes before triggering scan")
|
||||
log.Debug(ctx, "Watcher: Detected changes. Waiting for more changes before triggering scan",
|
||||
"libraryID", lib.ID, "name", lib.Name, "path", lib.Path)
|
||||
waiting = true
|
||||
}
|
||||
|
||||
trigger.Reset(w.triggerWait)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func watchLib(ctx context.Context, lib model.Library, watchChan chan struct{}) {
|
||||
func (w *watcher) Watch(ctx context.Context, lib *model.Library) error {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
// Stop existing watcher if any
|
||||
if existingInstance, exists := w.libraryWatchers[lib.ID]; exists {
|
||||
log.Debug(ctx, "Stopping existing watcher before starting new one", "libraryID", lib.ID, "name", lib.Name)
|
||||
existingInstance.cancel()
|
||||
}
|
||||
|
||||
// Start new watcher
|
||||
watcherCtx, cancel := context.WithCancel(w.mainCtx)
|
||||
instance := &libraryWatcherInstance{
|
||||
library: lib,
|
||||
cancel: cancel,
|
||||
}
|
||||
|
||||
w.libraryWatchers[lib.ID] = instance
|
||||
|
||||
// Start watching in a goroutine
|
||||
go func() {
|
||||
defer func() {
|
||||
w.mu.Lock()
|
||||
if currentInstance, exists := w.libraryWatchers[lib.ID]; exists && currentInstance == instance {
|
||||
delete(w.libraryWatchers, lib.ID)
|
||||
}
|
||||
w.mu.Unlock()
|
||||
}()
|
||||
|
||||
err := w.watchLibrary(watcherCtx, lib)
|
||||
if err != nil && watcherCtx.Err() == nil { // Only log error if not due to cancellation
|
||||
log.Error(ctx, "Watcher error", "libraryID", lib.ID, "name", lib.Name, "path", lib.Path, err)
|
||||
}
|
||||
}()
|
||||
|
||||
log.Info(ctx, "Started watcher for library", "libraryID", lib.ID, "name", lib.Name, "path", lib.Path)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *watcher) StopWatching(ctx context.Context, libraryID int) error {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
instance, exists := w.libraryWatchers[libraryID]
|
||||
if !exists {
|
||||
log.Debug(ctx, "No watcher found to stop", "libraryID", libraryID)
|
||||
return nil
|
||||
}
|
||||
|
||||
instance.cancel()
|
||||
delete(w.libraryWatchers, libraryID)
|
||||
|
||||
log.Info(ctx, "Stopped watcher for library", "libraryID", libraryID, "name", instance.library.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// watchLibrary implements the core watching logic for a single library (extracted from old watchLib function)
|
||||
func (w *watcher) watchLibrary(ctx context.Context, lib *model.Library) error {
|
||||
s, err := storage.For(lib.Path)
|
||||
if err != nil {
|
||||
log.Error(ctx, "Watcher: Error creating storage", "library", lib.ID, "path", lib.Path, err)
|
||||
return
|
||||
return fmt.Errorf("creating storage: %w", err)
|
||||
}
|
||||
|
||||
fsys, err := s.FS()
|
||||
if err != nil {
|
||||
log.Error(ctx, "Watcher: Error getting FS", "library", lib.ID, "path", lib.Path, err)
|
||||
return
|
||||
return fmt.Errorf("getting FS: %w", err)
|
||||
}
|
||||
|
||||
watcher, ok := s.(storage.Watcher)
|
||||
if !ok {
|
||||
log.Info(ctx, "Watcher not supported", "library", lib.ID, "path", lib.Path)
|
||||
return
|
||||
log.Info(ctx, "Watcher not supported for storage type", "libraryID", lib.ID, "path", lib.Path)
|
||||
return nil
|
||||
}
|
||||
|
||||
c, err := watcher.Start(ctx)
|
||||
if err != nil {
|
||||
log.Error(ctx, "Watcher: Error watching library", "library", lib.ID, "path", lib.Path, err)
|
||||
return
|
||||
return fmt.Errorf("starting watcher: %w", err)
|
||||
}
|
||||
|
||||
absLibPath, err := filepath.Abs(lib.Path)
|
||||
if err != nil {
|
||||
log.Error(ctx, "Watcher: Error converting lib.Path to absolute", "library", lib.ID, "path", lib.Path, err)
|
||||
return
|
||||
return fmt.Errorf("converting to absolute path: %w", err)
|
||||
}
|
||||
log.Info(ctx, "Watcher started", "library", lib.ID, "libPath", lib.Path, "absoluteLibPath", absLibPath)
|
||||
|
||||
log.Info(ctx, "Watcher started for library", "libraryID", lib.ID, "name", lib.Name, "path", lib.Path, "absoluteLibPath", absLibPath)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
log.Debug(ctx, "Watcher stopped due to context cancellation", "libraryID", lib.ID, "name", lib.Name)
|
||||
return nil
|
||||
case path := <-c:
|
||||
path, err = filepath.Rel(absLibPath, path)
|
||||
if err != nil {
|
||||
log.Error(ctx, "Watcher: Error getting relative path", "library", lib.ID, "libPath", absLibPath, "path", path, err)
|
||||
log.Error(ctx, "Error getting relative path", "libraryID", lib.ID, "absolutePath", absLibPath, "path", path, err)
|
||||
continue
|
||||
}
|
||||
|
||||
if isIgnoredPath(ctx, fsys, path) {
|
||||
log.Trace(ctx, "Watcher: Ignoring change", "library", lib.ID, "path", path)
|
||||
log.Trace(ctx, "Ignoring change", "libraryID", lib.ID, "path", path)
|
||||
continue
|
||||
}
|
||||
log.Trace(ctx, "Watcher: Detected change", "library", lib.ID, "path", path, "libPath", absLibPath)
|
||||
watchChan <- struct{}{}
|
||||
|
||||
log.Trace(ctx, "Detected change", "libraryID", lib.ID, "path", path, "absoluteLibPath", absLibPath)
|
||||
|
||||
// Notify the main watcher of changes
|
||||
select {
|
||||
case w.watcherNotify <- *lib:
|
||||
default:
|
||||
// Channel is full, notification already pending
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue