From e769ddf76c7b321784b108dcd628c00163bd58fc Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 28 Dec 2025 13:30:55 -0500 Subject: [PATCH] refactor(plugins): break manager.go into smaller, focused files Signed-off-by: Deluan --- plugins/README.md | 2 +- plugins/cmd/hostgen/integration_test.go | 16 +- plugins/host_scheduler_test.go | 2 +- plugins/host_subsonicapi_test.go | 2 +- plugins/manager.go | 766 +----------------- plugins/manager_cache.go | 92 +++ plugins/manager_cache_test.go | 187 +++++ plugins/manager_call.go | 87 ++ plugins/manager_loader.go | 367 +++++++++ plugins/manager_plugin.go | 43 + plugins/manager_sync.go | 226 ++++++ plugins/manager_test.go | 185 +---- plugins/{watcher.go => manager_watcher.go} | 8 +- ...atcher_test.go => manager_watcher_test.go} | 4 +- 14 files changed, 1028 insertions(+), 959 deletions(-) create mode 100644 plugins/manager_cache.go create mode 100644 plugins/manager_cache_test.go create mode 100644 plugins/manager_call.go create mode 100644 plugins/manager_loader.go create mode 100644 plugins/manager_plugin.go create mode 100644 plugins/manager_sync.go rename plugins/{watcher.go => manager_watcher.go} (97%) rename plugins/{watcher_test.go => manager_watcher_test.go} (98%) diff --git a/plugins/README.md b/plugins/README.md index e7cb38971..731867b66 100644 --- a/plugins/README.md +++ b/plugins/README.md @@ -520,7 +520,7 @@ manager := plugins.GetManager() err := manager.LoadPlugin("my-plugin") // Unload a running plugin -err := manager.UnloadPlugin("my-plugin") +err := manager.unloadPlugin("my-plugin") // Reload a plugin (unload + load) err := manager.ReloadPlugin("my-plugin") diff --git a/plugins/cmd/hostgen/integration_test.go b/plugins/cmd/hostgen/integration_test.go index 2730f0470..52bb80dff 100644 --- a/plugins/cmd/hostgen/integration_test.go +++ b/plugins/cmd/hostgen/integration_test.go @@ -10,14 +10,6 @@ import ( . "github.com/onsi/gomega" ) -var testdataDir string - -func readTestdata(filename string) string { - content, err := os.ReadFile(filepath.Join(testdataDir, filename)) - Expect(err).ToNot(HaveOccurred(), "Failed to read testdata file: %s", filename) - return string(content) -} - var _ = Describe("hostgen CLI", Ordered, func() { var ( testDir string @@ -527,6 +519,14 @@ type TestService interface { }) }) +var testdataDir string + +func readTestdata(filename string) string { + content, err := os.ReadFile(filepath.Join(testdataDir, filename)) + Expect(err).ToNot(HaveOccurred(), "Failed to read testdata file: %s", filename) + return string(content) +} + func mustGetWd(t FullGinkgoTInterface) string { dir, err := os.Getwd() if err != nil { diff --git a/plugins/host_scheduler_test.go b/plugins/host_scheduler_test.go index d32980ae1..24e56635e 100644 --- a/plugins/host_scheduler_test.go +++ b/plugins/host_scheduler_test.go @@ -300,7 +300,7 @@ var _ = Describe("SchedulerService", Ordered, func() { Expect(mockSched.GetCallbackCount()).To(Equal(1)) // Only recurring task uses scheduler Expect(mockTimers.GetTimerCount()).To(Equal(1)) // Only one-time task uses timer - err = manager.UnloadPlugin("test-scheduler") + err = manager.unloadPlugin("test-scheduler") Expect(err).ToNot(HaveOccurred()) Expect(findSchedulerService(manager, "test-scheduler")).To(BeNil()) diff --git a/plugins/host_subsonicapi_test.go b/plugins/host_subsonicapi_test.go index 30d5117be..e2ec8260a 100644 --- a/plugins/host_subsonicapi_test.go +++ b/plugins/host_subsonicapi_test.go @@ -72,7 +72,7 @@ var _ = Describe("SubsonicAPI Host Function", Ordered, func() { manager.SetSubsonicRouter(router) // Pre-enable the plugin in the mock repo so it loads on startup - // Compute SHA256 of the plugin file to match what SyncPlugins will compute + // Compute SHA256 of the plugin file to match what syncPlugins will compute pluginPath := filepath.Join(tmpDir, "test-subsonicapi-plugin.wasm") wasmData, err := os.ReadFile(pluginPath) Expect(err).ToNot(HaveOccurred()) diff --git a/plugins/manager.go b/plugins/manager.go index 0b0f0f486..545f013db 100644 --- a/plugins/manager.go +++ b/plugins/manager.go @@ -1,40 +1,25 @@ package plugins import ( - "cmp" "context" - "crypto/rand" - "crypto/sha256" - "encoding/hex" - "encoding/json" - "errors" "fmt" - "io" - "io/fs" "net/http" "os" "path/filepath" "runtime" - "slices" - "strings" "sync" "sync/atomic" "time" - "github.com/dustin/go-humanize" extism "github.com/extism/go-sdk" "github.com/navidrome/navidrome/conf" "github.com/navidrome/navidrome/core/agents" "github.com/navidrome/navidrome/core/scrobbler" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" - "github.com/navidrome/navidrome/model/request" - "github.com/navidrome/navidrome/plugins/host" - "github.com/navidrome/navidrome/scheduler" "github.com/navidrome/navidrome/utils/singleton" "github.com/rjeczalik/notify" "github.com/tetratelabs/wazero" - "golang.org/x/sync/errgroup" ) const ( @@ -75,38 +60,6 @@ type Manager struct { ds model.DataStore } -// plugin represents a loaded plugin -type plugin struct { - name string // Plugin name (from filename) - path string // Path to the wasm file - manifest *Manifest - compiled *extism.CompiledPlugin - capabilities []Capability // Auto-detected capabilities based on exported functions - closers []io.Closer // Cleanup functions to call on unload -} - -func (p *plugin) instance() (*extism.Plugin, error) { - instance, err := p.compiled.Instance(context.Background(), extism.PluginInstanceConfig{ - ModuleConfig: wazero.NewModuleConfig().WithSysWalltime().WithRandSource(rand.Reader), - }) - if err != nil { - return nil, err - } - instance.SetLogger(extismLogger(p.name)) - return instance, nil -} - -func (p *plugin) Close() error { - var errs []error - for _, f := range p.closers { - err := f.Close() - if err != nil { - errs = append(errs, err) - } - } - return errors.Join(errs...) -} - // GetManager returns a singleton instance of the plugin manager. // The manager is not started automatically; call Start() to begin loading plugins. func GetManager(ds model.DataStore) *Manager { @@ -175,7 +128,7 @@ func (m *Manager) Start(ctx context.Context) error { log.Info(ctx, "Starting plugin manager", "folder", folder) // Sync plugins folder with DB - if err := m.SyncPlugins(ctx, folder); err != nil { + if err := m.syncPlugins(ctx, folder); err != nil { log.Error(ctx, "Error syncing plugins with DB", err) // Continue - we can still try to load plugins } @@ -316,560 +269,6 @@ func (m *Manager) GetPluginInfo() map[string]PluginInfo { return info } -// adminContext returns a context with admin privileges for DB operations. -func adminContext(ctx context.Context) context.Context { - return request.WithUser(ctx, model.User{IsAdmin: true}) -} - -// marshalManifest marshals a manifest to JSON string, returning empty string on error. -func marshalManifest(m *Manifest) string { - b, _ := json.Marshal(m) - return string(b) -} - -// addPluginToDB adds a new plugin to the database as disabled. -func (m *Manager) addPluginToDB(ctx context.Context, repo model.PluginRepository, name, path string, metadata *PluginMetadata) error { - now := time.Now() - newPlugin := &model.Plugin{ - ID: name, - Path: path, - Manifest: marshalManifest(metadata.Manifest), - SHA256: metadata.SHA256, - Enabled: false, - CreatedAt: now, - UpdatedAt: now, - } - if err := repo.Put(newPlugin); err != nil { - return fmt.Errorf("adding plugin to DB: %w", err) - } - log.Info(ctx, "Discovered new plugin", "plugin", name) - return nil -} - -// updatePluginInDB updates an existing plugin in the database after a file change. -// If the plugin was enabled, it will be unloaded and disabled. -func (m *Manager) updatePluginInDB(ctx context.Context, repo model.PluginRepository, dbPlugin *model.Plugin, path string, metadata *PluginMetadata) error { - wasEnabled := dbPlugin.Enabled - if wasEnabled { - if err := m.UnloadPlugin(dbPlugin.ID); err != nil { - log.Debug(ctx, "Plugin not loaded during change", "plugin", dbPlugin.ID, err) - } - } - dbPlugin.Path = path - dbPlugin.Manifest = marshalManifest(metadata.Manifest) - dbPlugin.SHA256 = metadata.SHA256 - dbPlugin.Enabled = false - dbPlugin.LastError = "" - dbPlugin.UpdatedAt = time.Now() - if err := repo.Put(dbPlugin); err != nil { - return fmt.Errorf("updating plugin in DB: %w", err) - } - log.Info(ctx, "Plugin file changed", "plugin", dbPlugin.ID, "wasEnabled", wasEnabled) - return nil -} - -// removePluginFromDB removes a plugin from the database. -// If the plugin was enabled, it will be unloaded first. -func (m *Manager) removePluginFromDB(ctx context.Context, repo model.PluginRepository, dbPlugin *model.Plugin) error { - if dbPlugin.Enabled { - if err := m.UnloadPlugin(dbPlugin.ID); err != nil { - log.Debug(ctx, "Plugin not loaded during removal", "plugin", dbPlugin.ID, err) - } - } - if err := repo.Delete(dbPlugin.ID); err != nil { - return fmt.Errorf("deleting plugin from DB: %w", err) - } - log.Info(ctx, "Plugin removed", "plugin", dbPlugin.ID) - return nil -} - -// PluginMetadata holds the extracted information from a plugin file -// without fully initializing the plugin. -type PluginMetadata struct { - Manifest *Manifest - SHA256 string -} - -// computeFileSHA256 computes the SHA-256 hash of a file without loading it into memory. -// This is used for quick change detection before full plugin compilation. -func computeFileSHA256(path string) (string, error) { - f, err := os.Open(path) - if err != nil { - return "", err - } - defer f.Close() - - h := sha256.New() - if _, err := io.Copy(h, f); err != nil { - return "", err - } - return hex.EncodeToString(h.Sum(nil)), nil -} - -// compiledPluginInfo holds the intermediate compilation result used by both -// ExtractManifest and loadPluginWithConfig. -type compiledPluginInfo struct { - wasmBytes []byte - sha256 string - manifest *Manifest - compiled *extism.CompiledPlugin -} - -// serviceContext provides dependencies needed by host service factories. -type serviceContext struct { - pluginName string - manager *Manager - permissions *Permissions -} - -// hostServiceEntry defines a host service for table-driven registration. -type hostServiceEntry struct { - name string - hasPermission func(*Permissions) bool - registerStubs func() []extism.HostFunction - create func(*serviceContext) ([]extism.HostFunction, io.Closer) -} - -// hostServices defines all available host services. -// Adding a new host service only requires adding an entry here. -var hostServices = []hostServiceEntry{ - { - name: "SubsonicAPI", - hasPermission: func(p *Permissions) bool { return p != nil && p.Subsonicapi != nil }, - registerStubs: func() []extism.HostFunction { return host.RegisterSubsonicAPIHostFunctions(nil) }, - create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { - perm := ctx.permissions.Subsonicapi - service := newSubsonicAPIService(ctx.pluginName, ctx.manager.subsonicRouter, ctx.manager.ds, perm) - return host.RegisterSubsonicAPIHostFunctions(service), nil - }, - }, - { - name: "Scheduler", - hasPermission: func(p *Permissions) bool { return p != nil && p.Scheduler != nil }, - registerStubs: func() []extism.HostFunction { return host.RegisterSchedulerHostFunctions(nil) }, - create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { - service := newSchedulerService(ctx.pluginName, ctx.manager, scheduler.GetInstance()) - return host.RegisterSchedulerHostFunctions(service), service - }, - }, - { - name: "WebSocket", - hasPermission: func(p *Permissions) bool { return p != nil && p.Websocket != nil }, - registerStubs: func() []extism.HostFunction { return host.RegisterWebSocketHostFunctions(nil) }, - create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { - perm := ctx.permissions.Websocket - service := newWebSocketService(ctx.pluginName, ctx.manager, perm.AllowedHosts) - return host.RegisterWebSocketHostFunctions(service), service - }, - }, - { - name: "Artwork", - hasPermission: func(p *Permissions) bool { return p != nil && p.Artwork != nil }, - registerStubs: func() []extism.HostFunction { return host.RegisterArtworkHostFunctions(nil) }, - create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { - service := newArtworkService() - return host.RegisterArtworkHostFunctions(service), nil - }, - }, - { - name: "Cache", - hasPermission: func(p *Permissions) bool { return p != nil && p.Cache != nil }, - registerStubs: func() []extism.HostFunction { return host.RegisterCacheHostFunctions(nil) }, - create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { - service := newCacheService(ctx.pluginName) - return host.RegisterCacheHostFunctions(service), service - }, - }, -} - -// stubHostFunctions returns the list of stub host functions needed for initial plugin compilation. -func stubHostFunctions() []extism.HostFunction { - var stubs []extism.HostFunction - for _, entry := range hostServices { - stubs = append(stubs, entry.registerStubs()...) - } - return stubs -} - -// compileAndExtractManifest reads a wasm file, compiles it with cache, and extracts the manifest. -// The caller is responsible for closing the returned compiled plugin when done. -func (m *Manager) compileAndExtractManifest(ctx context.Context, wasmPath string, config map[string]string) (*compiledPluginInfo, error) { - wasmBytes, err := os.ReadFile(wasmPath) - if err != nil { - return nil, fmt.Errorf("reading wasm file: %w", err) - } - - // Compute SHA-256 hash - hash := sha256.Sum256(wasmBytes) - hashHex := hex.EncodeToString(hash[:]) - - // Extract plugin name from path for logging - pluginName := strings.TrimSuffix(filepath.Base(wasmPath), ".wasm") - - pluginManifest := extism.Manifest{ - Wasm: []extism.Wasm{ - extism.WasmData{Data: wasmBytes, Name: "main"}, - }, - Config: config, - Timeout: uint64(defaultTimeout.Milliseconds()), - } - extismConfig := extism.PluginConfig{ - EnableWasi: true, - RuntimeConfig: wazero.NewRuntimeConfig().WithCompilationCache(m.cache), - } - - compiled, err := extism.NewCompiledPlugin(ctx, pluginManifest, extismConfig, stubHostFunctions()) - if err != nil { - return nil, fmt.Errorf("compiling plugin: %w", err) - } - - instance, err := compiled.Instance(ctx, extism.PluginInstanceConfig{}) - if err != nil { - compiled.Close(ctx) - return nil, fmt.Errorf("creating instance: %w", err) - } - defer instance.Close(ctx) - instance.SetLogger(extismLogger(pluginName)) - - exit, manifestBytes, err := instance.Call(manifestFunction, nil) - if err != nil { - compiled.Close(ctx) - return nil, fmt.Errorf("calling manifest function: %w", err) - } - if exit != 0 { - compiled.Close(ctx) - return nil, fmt.Errorf("manifest function exited with code %d", exit) - } - - var manifest Manifest - if err := json.Unmarshal(manifestBytes, &manifest); err != nil { - compiled.Close(ctx) - return nil, fmt.Errorf("parsing manifest: %w", err) - } - - return &compiledPluginInfo{ - wasmBytes: wasmBytes, - sha256: hashHex, - manifest: &manifest, - compiled: compiled, - }, nil -} - -// ExtractManifest loads a wasm file, computes its SHA-256 hash, extracts the manifest, -// and immediately closes without full plugin initialization. -// This is a lightweight operation used for plugin discovery and change detection. -// The compilation is cached to speed up subsequent EnablePlugin calls. -func (m *Manager) ExtractManifest(wasmPath string) (*PluginMetadata, error) { - if m.stopped.Load() { - return nil, fmt.Errorf("manager is stopped") - } - - info, err := m.compileAndExtractManifest(context.Background(), wasmPath, nil) - if err != nil { - return nil, err - } - defer info.compiled.Close(context.Background()) - - return &PluginMetadata{ - Manifest: info.manifest, - SHA256: info.sha256, - }, nil -} - -// SyncPlugins scans the plugins folder and synchronizes with the database. -// It handles new, changed, and removed plugins by comparing SHA-256 hashes. -// - New plugins are added to DB as disabled -// - Changed plugins are updated in DB and disabled if they were enabled -// - Removed plugins are deleted from DB (after unloading if enabled) -func (m *Manager) SyncPlugins(ctx context.Context, folder string) error { - if m.ds == nil { - return fmt.Errorf("datastore not configured") - } - - adminCtx := adminContext(ctx) - - // Read current plugins from folder - entries, err := os.ReadDir(folder) - if err != nil { - if os.IsNotExist(err) { - log.Debug(ctx, "Plugins folder does not exist", "folder", folder) - return nil - } - return fmt.Errorf("reading plugins folder: %w", err) - } - - // Build map of files in folder - filesOnDisk := make(map[string]string) // name -> path - for _, entry := range entries { - if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".wasm") { - continue - } - name := strings.TrimSuffix(entry.Name(), ".wasm") - filesOnDisk[name] = filepath.Join(folder, entry.Name()) - } - - // Get all plugins from DB - repo := m.ds.Plugin(adminCtx) - dbPlugins, err := repo.GetAll() - if err != nil { - return fmt.Errorf("reading plugins from DB: %w", err) - } - pluginsInDB := make(map[string]*model.Plugin) - for i := range dbPlugins { - pluginsInDB[dbPlugins[i].ID] = &dbPlugins[i] - } - - now := time.Now() - - // Process files on disk - for name, path := range filesOnDisk { - dbPlugin, exists := pluginsInDB[name] - - // Compute SHA256 first (lightweight operation) to check if plugin changed - sha256Hash, err := computeFileSHA256(path) - if err != nil { - log.Error(ctx, "Failed to compute SHA256 for plugin", "plugin", name, "path", path, err) - continue - } - - // If plugin exists in DB with same hash, skip full manifest extraction - if exists && dbPlugin.SHA256 == sha256Hash { - // Plugin unchanged - just update path in case folder moved - if dbPlugin.Path != path { - dbPlugin.Path = path - dbPlugin.UpdatedAt = now - if err := repo.Put(dbPlugin); err != nil { - log.Error(ctx, "Failed to update plugin path in DB", "plugin", name, err) - } - } - delete(pluginsInDB, name) - continue - } - - // Plugin is new or changed - need full manifest extraction - metadata, err := m.ExtractManifest(path) - if err != nil { - log.Error(ctx, "Failed to extract manifest from plugin", "plugin", name, "path", path, err) - // Store error in DB if plugin exists - if exists { - dbPlugin.LastError = err.Error() - dbPlugin.UpdatedAt = now - if dbPlugin.Enabled { - // Unload broken plugin - if unloadErr := m.UnloadPlugin(name); unloadErr != nil { - log.Debug(ctx, "Plugin not loaded", "plugin", name) - } - dbPlugin.Enabled = false - } - if putErr := repo.Put(dbPlugin); putErr != nil { - log.Error(ctx, "Failed to update plugin in DB", "plugin", name, err) - } - } - delete(pluginsInDB, name) - continue - } - - if !exists { - // New plugin - add to DB as disabled - if err := m.addPluginToDB(ctx, repo, name, path, metadata); err != nil { - log.Error(ctx, "Failed to add plugin to DB", "plugin", name, err) - } - } else { - // Plugin changed - update DB - if err := m.updatePluginInDB(ctx, repo, dbPlugin, path, metadata); err != nil { - log.Error(ctx, "Failed to update plugin in DB", "plugin", name, err) - } - } - // Mark as processed - delete(pluginsInDB, name) - } - - // Remove plugins no longer on disk - for _, dbPlugin := range pluginsInDB { - if err := m.removePluginFromDB(ctx, repo, dbPlugin); err != nil { - log.Error(ctx, "Failed to delete plugin from DB", "plugin", dbPlugin.ID, err) - } - } - - return nil -} - -// loadEnabledPlugins loads all enabled plugins from the database. -func (m *Manager) loadEnabledPlugins(ctx context.Context) error { - if m.ds == nil { - return fmt.Errorf("datastore not configured") - } - - adminCtx := adminContext(ctx) - repo := m.ds.Plugin(adminCtx) - - plugins, err := repo.GetAll() - if err != nil { - return fmt.Errorf("reading plugins from DB: %w", err) - } - - g := errgroup.Group{} - g.SetLimit(maxPluginLoadConcurrency) - - for _, p := range plugins { - if !p.Enabled { - continue - } - - plugin := p // Capture for goroutine - g.Go(func() error { - start := time.Now() - log.Debug(ctx, "Loading enabled plugin", "plugin", plugin.ID, "path", plugin.Path) - - // Panic recovery - defer func() { - if r := recover(); r != nil { - log.Error(ctx, "Panic while loading plugin", "plugin", plugin.ID, "panic", r) - } - }() - - if err := m.loadPluginWithConfig(plugin.ID, plugin.Path, plugin.Config); err != nil { - // Store error in DB - plugin.LastError = err.Error() - plugin.Enabled = false - plugin.UpdatedAt = time.Now() - if putErr := repo.Put(&plugin); putErr != nil { - log.Error(ctx, "Failed to update plugin error in DB", "plugin", plugin.ID, putErr) - } - log.Error(ctx, "Failed to load plugin", "plugin", plugin.ID, err) - return nil - } - - // Clear any previous error - if plugin.LastError != "" { - plugin.LastError = "" - plugin.UpdatedAt = time.Now() - if putErr := repo.Put(&plugin); putErr != nil { - log.Error(ctx, "Failed to clear plugin error in DB", "plugin", plugin.ID, putErr) - } - } - - m.mu.RLock() - loadedPlugin := m.plugins[plugin.ID] - m.mu.RUnlock() - if loadedPlugin != nil { - log.Info(ctx, "Loaded plugin", "plugin", plugin.ID, "manifest", loadedPlugin.manifest.Name, - "capabilities", loadedPlugin.capabilities, "duration", time.Since(start)) - } - return nil - }) - } - - return g.Wait() -} - -// loadPluginWithConfig loads a plugin with configuration from DB. -func (m *Manager) loadPluginWithConfig(name, wasmPath, configJSON string) error { - if m.stopped.Load() { - return fmt.Errorf("manager is stopped") - } - - // Track this operation - m.loadWg.Add(1) - defer m.loadWg.Done() - - if m.stopped.Load() { - return fmt.Errorf("manager is stopped") - } - - // Parse config from JSON - var pluginConfig map[string]string - if configJSON != "" { - if err := json.Unmarshal([]byte(configJSON), &pluginConfig); err != nil { - return fmt.Errorf("parsing plugin config: %w", err) - } - } - - // Compile and extract manifest using shared helper - info, err := m.compileAndExtractManifest(m.ctx, wasmPath, pluginConfig) - if err != nil { - return err - } - - // Create instance to detect capabilities - instance, err := info.compiled.Instance(m.ctx, extism.PluginInstanceConfig{}) - if err != nil { - info.compiled.Close(m.ctx) - return fmt.Errorf("creating instance: %w", err) - } - instance.SetLogger(extismLogger(name)) - capabilities := detectCapabilities(instance) - instance.Close(m.ctx) - - // Build host functions based on permissions - var hostFunctions []extism.HostFunction - var closers []io.Closer - - // Build extism manifest for potential recompilation - pluginManifest := extism.Manifest{ - Wasm: []extism.Wasm{ - extism.WasmData{Data: info.wasmBytes, Name: "main"}, - }, - Config: pluginConfig, - Timeout: uint64(defaultTimeout.Milliseconds()), - } - - if hosts := info.manifest.AllowedHosts(); len(hosts) > 0 { - pluginManifest.AllowedHosts = hosts - } - - // Register host functions based on permissions using table-driven approach - svcCtx := &serviceContext{ - pluginName: name, - manager: m, - permissions: info.manifest.Permissions, - } - for _, entry := range hostServices { - if entry.hasPermission(info.manifest.Permissions) { - funcs, closer := entry.create(svcCtx) - hostFunctions = append(hostFunctions, funcs...) - if closer != nil { - closers = append(closers, closer) - } - } - } - - // Check if the plugin needs to be recompiled with real host functions - compiled := info.compiled - needsRecompile := len(pluginManifest.AllowedHosts) > 0 || len(hostFunctions) > 0 - - // Recompile if needed. It is actually not a "recompile" since the first compilation - // should be cached by wazero. We just need to do it this way to provide the real host functions. - if needsRecompile { - log.Trace(m.ctx, "Recompiling plugin with host functions", "plugin", name) - info.compiled.Close(m.ctx) - extismConfig := extism.PluginConfig{ - EnableWasi: true, - RuntimeConfig: wazero.NewRuntimeConfig().WithCompilationCache(m.cache), - } - compiled, err = extism.NewCompiledPlugin(m.ctx, pluginManifest, extismConfig, hostFunctions) - if err != nil { - return err - } - } - - m.mu.Lock() - m.plugins[name] = &plugin{ - name: name, - path: wasmPath, - manifest: info.manifest, - compiled: compiled, - capabilities: capabilities, - closers: closers, - } - m.mu.Unlock() - - // Call plugin init function - callPluginInit(m.ctx, m.plugins[name]) - - return nil -} - // EnablePlugin enables a plugin by loading it and updating the DB. // Returns an error if the plugin is not found in DB or fails to load. func (m *Manager) EnablePlugin(ctx context.Context, id string) error { @@ -904,7 +303,7 @@ func (m *Manager) EnablePlugin(ctx context.Context, id string) error { plugin.UpdatedAt = time.Now() if err := repo.Put(plugin); err != nil { // Unload since we couldn't update DB - _ = m.UnloadPlugin(id) + _ = m.unloadPlugin(id) return fmt.Errorf("updating plugin in DB: %w", err) } @@ -932,7 +331,7 @@ func (m *Manager) DisablePlugin(ctx context.Context, id string) error { } // Unload the plugin - if err := m.UnloadPlugin(id); err != nil { + if err := m.unloadPlugin(id); err != nil { log.Debug(ctx, "Plugin was not loaded", "plugin", id) } @@ -973,7 +372,7 @@ func (m *Manager) UpdatePluginConfig(ctx context.Context, id, configJSON string) // Reload if enabled if wasEnabled { - if err := m.UnloadPlugin(id); err != nil { + if err := m.unloadPlugin(id); err != nil { log.Debug(ctx, "Plugin was not loaded", "plugin", id) } if err := m.loadPluginWithConfig(plugin.ID, plugin.Path, configJSON); err != nil { @@ -988,9 +387,9 @@ func (m *Manager) UpdatePluginConfig(ctx context.Context, id, configJSON string) return nil } -// UnloadPlugin removes a plugin from the manager and closes its resources. +// unloadPlugin removes a plugin from the manager and closes its resources. // Returns an error if the plugin is not found. -func (m *Manager) UnloadPlugin(name string) error { +func (m *Manager) unloadPlugin(name string) error { m.mu.Lock() plugin, ok := m.plugins[name] if !ok { @@ -1021,156 +420,3 @@ func (m *Manager) UnloadPlugin(name string) error { log.Info(m.ctx, "Unloaded plugin", "plugin", name) return nil } - -var errFunctionNotFound = errors.New("function not found") - -// callPluginFunction is a helper to call a plugin function with input and output types. -// It handles JSON marshalling/unmarshalling and error checking. -func callPluginFunction[I any, O any](ctx context.Context, plugin *plugin, funcName string, input I) (O, error) { - start := time.Now() - - var result O - - // Create plugin instance - p, err := plugin.instance() - if err != nil { - return result, fmt.Errorf("failed to create plugin: %w", err) - } - defer p.Close(ctx) - - if !p.FunctionExists(funcName) { - log.Trace(ctx, "Plugin function not found", "plugin", plugin.name, "function", funcName) - return result, fmt.Errorf("%w: %s", errFunctionNotFound, funcName) - } - - inputBytes, err := json.Marshal(input) - if err != nil { - return result, fmt.Errorf("failed to marshal input: %w", err) - } - - startCall := time.Now() - exit, output, err := p.Call(funcName, inputBytes) - if err != nil { - log.Trace(ctx, "Plugin call failed", "plugin", plugin.name, "function", funcName, "pluginDuration", time.Since(startCall), "navidromeDuration", startCall.Sub(start), err) - return result, fmt.Errorf("plugin call failed: %w", err) - } - if exit != 0 { - return result, fmt.Errorf("plugin call exited with code %d", exit) - } - - if len(output) > 0 { - err = json.Unmarshal(output, &result) - if err != nil { - log.Trace(ctx, "Plugin call failed", "plugin", plugin.name, "function", funcName, "pluginDuration", time.Since(startCall), "navidromeDuration", startCall.Sub(start), err) - } - } - - log.Trace(ctx, "Plugin call succeeded", "plugin", plugin.name, "function", funcName, "pluginDuration", time.Since(startCall), "navidromeDuration", startCall.Sub(start)) - return result, err -} - -// extismLogger is a helper to log messages from Extism plugins -func extismLogger(pluginName string) func(level extism.LogLevel, msg string) { - return func(level extism.LogLevel, msg string) { - if level == extism.LogLevelOff { - return - } - log.Log(log.ParseLogLevel(level.String()), msg, "plugin", pluginName) - } -} - -// toExtismLogLevel converts a Navidrome log level to an extism LogLevel -func toExtismLogLevel(level log.Level) extism.LogLevel { - switch level { - case log.LevelTrace: - return extism.LogLevelTrace - case log.LevelDebug: - return extism.LogLevelDebug - case log.LevelInfo: - return extism.LogLevelInfo - case log.LevelWarn: - return extism.LogLevelWarn - case log.LevelError, log.LevelFatal: - return extism.LogLevelError - default: - return extism.LogLevelInfo - } -} - -// purgeCacheBySize removes the oldest files in dir until its total size is -// lower than or equal to maxSize. maxSize should be a human-readable string -// like "10MB" or "200K". If parsing fails or maxSize is "0", the function is -// a no-op. -func purgeCacheBySize(ctx context.Context, dir, maxSize string) { - sizeLimit, err := humanize.ParseBytes(maxSize) - if err != nil || sizeLimit == 0 { - return - } - - type fileInfo struct { - path string - size uint64 - mod int64 - } - - var files []fileInfo - var total uint64 - - walk := func(path string, d fs.DirEntry, err error) error { - if err != nil { - log.Trace(ctx, "Failed to access plugin cache entry", "path", path, err) - return nil //nolint:nilerr - } - if d.IsDir() { - return nil - } - info, err := d.Info() - if err != nil { - log.Trace(ctx, "Failed to get file info for plugin cache entry", "path", path, err) - return nil //nolint:nilerr - } - files = append(files, fileInfo{ - path: path, - size: uint64(info.Size()), - mod: info.ModTime().UnixMilli(), - }) - total += uint64(info.Size()) - return nil - } - - if err := filepath.WalkDir(dir, walk); err != nil { - if !os.IsNotExist(err) { - log.Warn(ctx, "Failed to traverse plugin cache directory", "path", dir, err) - } - return - } - - log.Trace(ctx, "Current plugin cache size", "path", dir, "size", humanize.Bytes(total), "sizeLimit", humanize.Bytes(sizeLimit)) - if total <= sizeLimit { - return - } - - log.Debug(ctx, "Purging plugin cache", "path", dir, "sizeLimit", humanize.Bytes(sizeLimit), "currentSize", humanize.Bytes(total)) - slices.SortFunc(files, func(i, j fileInfo) int { return cmp.Compare(i.mod, j.mod) }) - - for _, f := range files { - if total <= sizeLimit { - break - } - if err := os.Remove(f.path); err != nil { - log.Warn(ctx, "Failed to remove plugin cache entry", "path", f.path, "size", humanize.Bytes(f.size), err) - continue - } - total -= f.size - log.Debug(ctx, "Removed plugin cache entry", "path", f.path, "size", humanize.Bytes(f.size), "time", time.UnixMilli(f.mod), "remainingSize", humanize.Bytes(total)) - - // Remove empty parent directories - dirPath := filepath.Dir(f.path) - for dirPath != dir { - if err := os.Remove(dirPath); err != nil { - break - } - dirPath = filepath.Dir(dirPath) - } - } -} diff --git a/plugins/manager_cache.go b/plugins/manager_cache.go new file mode 100644 index 000000000..74b27171f --- /dev/null +++ b/plugins/manager_cache.go @@ -0,0 +1,92 @@ +package plugins + +import ( + "cmp" + "context" + "io/fs" + "os" + "path/filepath" + "slices" + "time" + + "github.com/dustin/go-humanize" + "github.com/navidrome/navidrome/log" +) + +// purgeCacheBySize removes the oldest files in dir until its total size is +// lower than or equal to maxSize. maxSize should be a human-readable string +// like "10MB" or "200K". If parsing fails or maxSize is "0", the function is +// a no-op. +func purgeCacheBySize(ctx context.Context, dir, maxSize string) { + sizeLimit, err := humanize.ParseBytes(maxSize) + if err != nil || sizeLimit == 0 { + return + } + + type fileInfo struct { + path string + size uint64 + mod int64 + } + + var files []fileInfo + var total uint64 + + walk := func(path string, d fs.DirEntry, err error) error { + if err != nil { + log.Trace(ctx, "Failed to access plugin cache entry", "path", path, err) + return nil //nolint:nilerr + } + if d.IsDir() { + return nil + } + info, err := d.Info() + if err != nil { + log.Trace(ctx, "Failed to get file info for plugin cache entry", "path", path, err) + return nil //nolint:nilerr + } + files = append(files, fileInfo{ + path: path, + size: uint64(info.Size()), + mod: info.ModTime().UnixMilli(), + }) + total += uint64(info.Size()) + return nil + } + + if err := filepath.WalkDir(dir, walk); err != nil { + if !os.IsNotExist(err) { + log.Warn(ctx, "Failed to traverse plugin cache directory", "path", dir, err) + } + return + } + + log.Trace(ctx, "Current plugin cache size", "path", dir, "size", humanize.Bytes(total), "sizeLimit", humanize.Bytes(sizeLimit)) + if total <= sizeLimit { + return + } + + log.Debug(ctx, "Purging plugin cache", "path", dir, "sizeLimit", humanize.Bytes(sizeLimit), "currentSize", humanize.Bytes(total)) + slices.SortFunc(files, func(i, j fileInfo) int { return cmp.Compare(i.mod, j.mod) }) + + for _, f := range files { + if total <= sizeLimit { + break + } + if err := os.Remove(f.path); err != nil { + log.Warn(ctx, "Failed to remove plugin cache entry", "path", f.path, "size", humanize.Bytes(f.size), err) + continue + } + total -= f.size + log.Debug(ctx, "Removed plugin cache entry", "path", f.path, "size", humanize.Bytes(f.size), "time", time.UnixMilli(f.mod), "remainingSize", humanize.Bytes(total)) + + // Remove empty parent directories + dirPath := filepath.Dir(f.path) + for dirPath != dir { + if err := os.Remove(dirPath); err != nil { + break + } + dirPath = filepath.Dir(dirPath) + } + } +} diff --git a/plugins/manager_cache_test.go b/plugins/manager_cache_test.go new file mode 100644 index 000000000..9411f767b --- /dev/null +++ b/plugins/manager_cache_test.go @@ -0,0 +1,187 @@ +package plugins + +import ( + "context" + "os" + "path/filepath" + "time" + + "github.com/dustin/go-humanize" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("purgeCacheBySize", func() { + var ( + tmpDir string + ctx context.Context + ) + + BeforeEach(func() { + var err error + ctx = GinkgoT().Context() + tmpDir, err = os.MkdirTemp("", "cache-purge-test-*") + Expect(err).ToNot(HaveOccurred()) + }) + + AfterEach(func() { + os.RemoveAll(tmpDir) + }) + + createFileWithSize := func(path string, sizeBytes int64, modTime time.Time) { + dir := filepath.Dir(path) + err := os.MkdirAll(dir, 0755) + Expect(err).ToNot(HaveOccurred()) + + f, err := os.Create(path) + Expect(err).ToNot(HaveOccurred()) + defer f.Close() + + // Write random data to reach desired size + if sizeBytes > 0 { + err = f.Truncate(sizeBytes) + Expect(err).ToNot(HaveOccurred()) + } + + // Set modification time + err = os.Chtimes(path, modTime, modTime) + Expect(err).ToNot(HaveOccurred()) + } + + getDirSize := func(dir string) uint64 { + var total uint64 + err := filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error { + if err != nil || d.IsDir() { + return nil + } + info, err := d.Info() + if err != nil { + return nil + } + total += uint64(info.Size()) + return nil + }) + Expect(err).ToNot(HaveOccurred()) + return total + } + + Context("when maxSize is invalid or zero", func() { + It("should not remove any files with invalid size", func() { + cacheDir := filepath.Join(tmpDir, "cache") + createFileWithSize(filepath.Join(cacheDir, "file1.bin"), 1000, time.Now()) + createFileWithSize(filepath.Join(cacheDir, "file2.bin"), 1000, time.Now()) + + purgeCacheBySize(ctx, cacheDir, "invalid") + + Expect(getDirSize(cacheDir)).To(Equal(uint64(2000))) + }) + + It("should not remove any files when maxSize is 0", func() { + cacheDir := filepath.Join(tmpDir, "cache") + createFileWithSize(filepath.Join(cacheDir, "file1.bin"), 1000, time.Now()) + createFileWithSize(filepath.Join(cacheDir, "file2.bin"), 1000, time.Now()) + + purgeCacheBySize(ctx, cacheDir, "0") + + Expect(getDirSize(cacheDir)).To(Equal(uint64(2000))) + }) + }) + + Context("when cache directory doesn't exist", func() { + It("should not error", func() { + nonExistentDir := filepath.Join(tmpDir, "nonexistent") + Expect(func() { + purgeCacheBySize(ctx, nonExistentDir, "100MB") + }).ToNot(Panic()) + }) + }) + + Context("when total size is under limit", func() { + It("should not remove any files", func() { + cacheDir := filepath.Join(tmpDir, "cache") + createFileWithSize(filepath.Join(cacheDir, "file1.bin"), 1000, time.Now()) + createFileWithSize(filepath.Join(cacheDir, "file2.bin"), 1000, time.Now()) + + purgeCacheBySize(ctx, cacheDir, "10KB") + + Expect(getDirSize(cacheDir)).To(Equal(uint64(2000))) + }) + }) + + Context("when total size exceeds limit", func() { + It("should remove oldest files first", func() { + cacheDir := filepath.Join(tmpDir, "cache") + now := time.Now() + + // Create files with different ages (1MB each) + oldestFile := filepath.Join(cacheDir, "old.bin") + middleFile := filepath.Join(cacheDir, "middle.bin") + newestFile := filepath.Join(cacheDir, "new.bin") + + createFileWithSize(oldestFile, 1*1024*1024, now.Add(-3*time.Hour)) + createFileWithSize(middleFile, 1*1024*1024, now.Add(-2*time.Hour)) + createFileWithSize(newestFile, 1*1024*1024, now.Add(-1*time.Hour)) + + // Set limit to 2MiB - should remove oldest file + purgeCacheBySize(ctx, cacheDir, "2MiB") + + // Oldest should be removed + _, err := os.Stat(oldestFile) + Expect(os.IsNotExist(err)).To(BeTrue(), "oldest file should be removed") + + // Others should remain + _, err = os.Stat(middleFile) + Expect(err).ToNot(HaveOccurred(), "middle file should remain") + + _, err = os.Stat(newestFile) + Expect(err).ToNot(HaveOccurred(), "newest file should remain") + }) + + It("should remove multiple files to get under limit", func() { + cacheDir := filepath.Join(tmpDir, "cache") + now := time.Now() + + // Create 5 files, 1MiB each (total 5MiB) + for i := 0; i < 5; i++ { + path := filepath.Join(cacheDir, filepath.Join("dir", "file"+string(rune('0'+i))+".bin")) + createFileWithSize(path, 1*1024*1024, now.Add(-time.Duration(5-i)*time.Hour)) + } + + // Set limit to 2.5MiB - should remove oldest 3 files (leaving 2MiB) + purgeCacheBySize(ctx, cacheDir, "2.5MiB") + + finalSize := getDirSize(cacheDir) + limit, _ := humanize.ParseBytes("2.5MiB") + Expect(finalSize).To(BeNumerically("<=", limit)) + }) + + It("should remove empty parent directories after removing files", func() { + cacheDir := filepath.Join(tmpDir, "cache") + now := time.Now() + + // Create files in subdirectories + oldFile := filepath.Join(cacheDir, "subdir1", "old.bin") + newFile := filepath.Join(cacheDir, "subdir2", "new.bin") + + createFileWithSize(oldFile, 2*1024*1024, now.Add(-2*time.Hour)) + createFileWithSize(newFile, 2*1024*1024, now.Add(-1*time.Hour)) + + // Set limit to 2MiB - should remove old file and its parent dir + purgeCacheBySize(ctx, cacheDir, "2MiB") + + // Old file and its parent dir should be removed + _, err := os.Stat(oldFile) + Expect(os.IsNotExist(err)).To(BeTrue()) + + _, err = os.Stat(filepath.Join(cacheDir, "subdir1")) + Expect(os.IsNotExist(err)).To(BeTrue(), "empty parent directory should be removed") + + // New file and its parent dir should remain + _, err = os.Stat(newFile) + Expect(err).ToNot(HaveOccurred()) + + _, err = os.Stat(filepath.Join(cacheDir, "subdir2")) + Expect(err).ToNot(HaveOccurred()) + }) + }) +}) diff --git a/plugins/manager_call.go b/plugins/manager_call.go new file mode 100644 index 000000000..1fa701d0a --- /dev/null +++ b/plugins/manager_call.go @@ -0,0 +1,87 @@ +package plugins + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + extism "github.com/extism/go-sdk" + "github.com/navidrome/navidrome/log" +) + +var errFunctionNotFound = errors.New("function not found") + +// callPluginFunction is a helper to call a plugin function with input and output types. +// It handles JSON marshalling/unmarshalling and error checking. +func callPluginFunction[I any, O any](ctx context.Context, plugin *plugin, funcName string, input I) (O, error) { + start := time.Now() + + var result O + + // Create plugin instance + p, err := plugin.instance() + if err != nil { + return result, fmt.Errorf("failed to create plugin: %w", err) + } + defer p.Close(ctx) + + if !p.FunctionExists(funcName) { + log.Trace(ctx, "Plugin function not found", "plugin", plugin.name, "function", funcName) + return result, fmt.Errorf("%w: %s", errFunctionNotFound, funcName) + } + + inputBytes, err := json.Marshal(input) + if err != nil { + return result, fmt.Errorf("failed to marshal input: %w", err) + } + + startCall := time.Now() + exit, output, err := p.Call(funcName, inputBytes) + if err != nil { + log.Trace(ctx, "Plugin call failed", "plugin", plugin.name, "function", funcName, "pluginDuration", time.Since(startCall), "navidromeDuration", startCall.Sub(start), err) + return result, fmt.Errorf("plugin call failed: %w", err) + } + if exit != 0 { + return result, fmt.Errorf("plugin call exited with code %d", exit) + } + + if len(output) > 0 { + err = json.Unmarshal(output, &result) + if err != nil { + log.Trace(ctx, "Plugin call failed", "plugin", plugin.name, "function", funcName, "pluginDuration", time.Since(startCall), "navidromeDuration", startCall.Sub(start), err) + } + } + + log.Trace(ctx, "Plugin call succeeded", "plugin", plugin.name, "function", funcName, "pluginDuration", time.Since(startCall), "navidromeDuration", startCall.Sub(start)) + return result, err +} + +// extismLogger is a helper to log messages from Extism plugins +func extismLogger(pluginName string) func(level extism.LogLevel, msg string) { + return func(level extism.LogLevel, msg string) { + if level == extism.LogLevelOff { + return + } + log.Log(log.ParseLogLevel(level.String()), msg, "plugin", pluginName) + } +} + +// toExtismLogLevel converts a Navidrome log level to an extism LogLevel +func toExtismLogLevel(level log.Level) extism.LogLevel { + switch level { + case log.LevelTrace: + return extism.LogLevelTrace + case log.LevelDebug: + return extism.LogLevelDebug + case log.LevelInfo: + return extism.LogLevelInfo + case log.LevelWarn: + return extism.LogLevelWarn + case log.LevelError, log.LevelFatal: + return extism.LogLevelError + default: + return extism.LogLevelInfo + } +} diff --git a/plugins/manager_loader.go b/plugins/manager_loader.go new file mode 100644 index 000000000..3d9aa99fb --- /dev/null +++ b/plugins/manager_loader.go @@ -0,0 +1,367 @@ +package plugins + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + extism "github.com/extism/go-sdk" + "github.com/navidrome/navidrome/log" + "github.com/navidrome/navidrome/plugins/host" + "github.com/navidrome/navidrome/scheduler" + "github.com/tetratelabs/wazero" + "golang.org/x/sync/errgroup" +) + +// serviceContext provides dependencies needed by host service factories. +type serviceContext struct { + pluginName string + manager *Manager + permissions *Permissions +} + +// hostServiceEntry defines a host service for table-driven registration. +type hostServiceEntry struct { + name string + hasPermission func(*Permissions) bool + registerStubs func() []extism.HostFunction + create func(*serviceContext) ([]extism.HostFunction, io.Closer) +} + +// hostServices defines all available host services. +// Adding a new host service only requires adding an entry here. +var hostServices = []hostServiceEntry{ + { + name: "SubsonicAPI", + hasPermission: func(p *Permissions) bool { return p != nil && p.Subsonicapi != nil }, + registerStubs: func() []extism.HostFunction { return host.RegisterSubsonicAPIHostFunctions(nil) }, + create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { + perm := ctx.permissions.Subsonicapi + service := newSubsonicAPIService(ctx.pluginName, ctx.manager.subsonicRouter, ctx.manager.ds, perm) + return host.RegisterSubsonicAPIHostFunctions(service), nil + }, + }, + { + name: "Scheduler", + hasPermission: func(p *Permissions) bool { return p != nil && p.Scheduler != nil }, + registerStubs: func() []extism.HostFunction { return host.RegisterSchedulerHostFunctions(nil) }, + create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { + service := newSchedulerService(ctx.pluginName, ctx.manager, scheduler.GetInstance()) + return host.RegisterSchedulerHostFunctions(service), service + }, + }, + { + name: "WebSocket", + hasPermission: func(p *Permissions) bool { return p != nil && p.Websocket != nil }, + registerStubs: func() []extism.HostFunction { return host.RegisterWebSocketHostFunctions(nil) }, + create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { + perm := ctx.permissions.Websocket + service := newWebSocketService(ctx.pluginName, ctx.manager, perm.AllowedHosts) + return host.RegisterWebSocketHostFunctions(service), service + }, + }, + { + name: "Artwork", + hasPermission: func(p *Permissions) bool { return p != nil && p.Artwork != nil }, + registerStubs: func() []extism.HostFunction { return host.RegisterArtworkHostFunctions(nil) }, + create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { + service := newArtworkService() + return host.RegisterArtworkHostFunctions(service), nil + }, + }, + { + name: "Cache", + hasPermission: func(p *Permissions) bool { return p != nil && p.Cache != nil }, + registerStubs: func() []extism.HostFunction { return host.RegisterCacheHostFunctions(nil) }, + create: func(ctx *serviceContext) ([]extism.HostFunction, io.Closer) { + service := newCacheService(ctx.pluginName) + return host.RegisterCacheHostFunctions(service), service + }, + }, +} + +// stubHostFunctions returns the list of stub host functions needed for initial plugin compilation. +func stubHostFunctions() []extism.HostFunction { + var stubs []extism.HostFunction + for _, entry := range hostServices { + stubs = append(stubs, entry.registerStubs()...) + } + return stubs +} + +// compiledPluginInfo holds the intermediate compilation result used by both +// extractManifest and loadPluginWithConfig. +type compiledPluginInfo struct { + wasmBytes []byte + sha256 string + manifest *Manifest + compiled *extism.CompiledPlugin +} + +// compileAndExtractManifest reads a wasm file, compiles it with cache, and extracts the manifest. +// The caller is responsible for closing the returned compiled plugin when done. +func (m *Manager) compileAndExtractManifest(ctx context.Context, wasmPath string, config map[string]string) (*compiledPluginInfo, error) { + wasmBytes, err := os.ReadFile(wasmPath) + if err != nil { + return nil, fmt.Errorf("reading wasm file: %w", err) + } + + // Compute SHA-256 hash + hash := sha256.Sum256(wasmBytes) + hashHex := hex.EncodeToString(hash[:]) + + // Extract plugin name from path for logging + pluginName := strings.TrimSuffix(filepath.Base(wasmPath), ".wasm") + + pluginManifest := extism.Manifest{ + Wasm: []extism.Wasm{ + extism.WasmData{Data: wasmBytes, Name: "main"}, + }, + Config: config, + Timeout: uint64(defaultTimeout.Milliseconds()), + } + extismConfig := extism.PluginConfig{ + EnableWasi: true, + RuntimeConfig: wazero.NewRuntimeConfig().WithCompilationCache(m.cache), + } + + compiled, err := extism.NewCompiledPlugin(ctx, pluginManifest, extismConfig, stubHostFunctions()) + if err != nil { + return nil, fmt.Errorf("compiling plugin: %w", err) + } + + instance, err := compiled.Instance(ctx, extism.PluginInstanceConfig{}) + if err != nil { + compiled.Close(ctx) + return nil, fmt.Errorf("creating instance: %w", err) + } + defer instance.Close(ctx) + instance.SetLogger(extismLogger(pluginName)) + + exit, manifestBytes, err := instance.Call(manifestFunction, nil) + if err != nil { + compiled.Close(ctx) + return nil, fmt.Errorf("calling manifest function: %w", err) + } + if exit != 0 { + compiled.Close(ctx) + return nil, fmt.Errorf("manifest function exited with code %d", exit) + } + + var manifest Manifest + if err := json.Unmarshal(manifestBytes, &manifest); err != nil { + compiled.Close(ctx) + return nil, fmt.Errorf("parsing manifest: %w", err) + } + + return &compiledPluginInfo{ + wasmBytes: wasmBytes, + sha256: hashHex, + manifest: &manifest, + compiled: compiled, + }, nil +} + +// extractManifest loads a wasm file, computes its SHA-256 hash, extracts the manifest, +// and immediately closes without full plugin initialization. +// This is a lightweight operation used for plugin discovery and change detection. +// The compilation is cached to speed up subsequent EnablePlugin calls. +func (m *Manager) extractManifest(wasmPath string) (*PluginMetadata, error) { + if m.stopped.Load() { + return nil, fmt.Errorf("manager is stopped") + } + + info, err := m.compileAndExtractManifest(context.Background(), wasmPath, nil) + if err != nil { + return nil, err + } + defer info.compiled.Close(context.Background()) + + return &PluginMetadata{ + Manifest: info.manifest, + SHA256: info.sha256, + }, nil +} + +// loadEnabledPlugins loads all enabled plugins from the database. +func (m *Manager) loadEnabledPlugins(ctx context.Context) error { + if m.ds == nil { + return fmt.Errorf("datastore not configured") + } + + adminCtx := adminContext(ctx) + repo := m.ds.Plugin(adminCtx) + + plugins, err := repo.GetAll() + if err != nil { + return fmt.Errorf("reading plugins from DB: %w", err) + } + + g := errgroup.Group{} + g.SetLimit(maxPluginLoadConcurrency) + + for _, p := range plugins { + if !p.Enabled { + continue + } + + plugin := p // Capture for goroutine + g.Go(func() error { + start := time.Now() + log.Debug(ctx, "Loading enabled plugin", "plugin", plugin.ID, "path", plugin.Path) + + // Panic recovery + defer func() { + if r := recover(); r != nil { + log.Error(ctx, "Panic while loading plugin", "plugin", plugin.ID, "panic", r) + } + }() + + if err := m.loadPluginWithConfig(plugin.ID, plugin.Path, plugin.Config); err != nil { + // Store error in DB + plugin.LastError = err.Error() + plugin.Enabled = false + plugin.UpdatedAt = time.Now() + if putErr := repo.Put(&plugin); putErr != nil { + log.Error(ctx, "Failed to update plugin error in DB", "plugin", plugin.ID, putErr) + } + log.Error(ctx, "Failed to load plugin", "plugin", plugin.ID, err) + return nil + } + + // Clear any previous error + if plugin.LastError != "" { + plugin.LastError = "" + plugin.UpdatedAt = time.Now() + if putErr := repo.Put(&plugin); putErr != nil { + log.Error(ctx, "Failed to clear plugin error in DB", "plugin", plugin.ID, putErr) + } + } + + m.mu.RLock() + loadedPlugin := m.plugins[plugin.ID] + m.mu.RUnlock() + if loadedPlugin != nil { + log.Info(ctx, "Loaded plugin", "plugin", plugin.ID, "manifest", loadedPlugin.manifest.Name, + "capabilities", loadedPlugin.capabilities, "duration", time.Since(start)) + } + return nil + }) + } + + return g.Wait() +} + +// loadPluginWithConfig loads a plugin with configuration from DB. +func (m *Manager) loadPluginWithConfig(name, wasmPath, configJSON string) error { + if m.stopped.Load() { + return fmt.Errorf("manager is stopped") + } + + // Track this operation + m.loadWg.Add(1) + defer m.loadWg.Done() + + if m.stopped.Load() { + return fmt.Errorf("manager is stopped") + } + + // Parse config from JSON + var pluginConfig map[string]string + if configJSON != "" { + if err := json.Unmarshal([]byte(configJSON), &pluginConfig); err != nil { + return fmt.Errorf("parsing plugin config: %w", err) + } + } + + // Compile and extract manifest using shared helper + info, err := m.compileAndExtractManifest(m.ctx, wasmPath, pluginConfig) + if err != nil { + return err + } + + // Create instance to detect capabilities + instance, err := info.compiled.Instance(m.ctx, extism.PluginInstanceConfig{}) + if err != nil { + info.compiled.Close(m.ctx) + return fmt.Errorf("creating instance: %w", err) + } + instance.SetLogger(extismLogger(name)) + capabilities := detectCapabilities(instance) + instance.Close(m.ctx) + + // Build host functions based on permissions + var hostFunctions []extism.HostFunction + var closers []io.Closer + + // Build extism manifest for potential recompilation + pluginManifest := extism.Manifest{ + Wasm: []extism.Wasm{ + extism.WasmData{Data: info.wasmBytes, Name: "main"}, + }, + Config: pluginConfig, + Timeout: uint64(defaultTimeout.Milliseconds()), + } + + if hosts := info.manifest.AllowedHosts(); len(hosts) > 0 { + pluginManifest.AllowedHosts = hosts + } + + // Register host functions based on permissions using table-driven approach + svcCtx := &serviceContext{ + pluginName: name, + manager: m, + permissions: info.manifest.Permissions, + } + for _, entry := range hostServices { + if entry.hasPermission(info.manifest.Permissions) { + funcs, closer := entry.create(svcCtx) + hostFunctions = append(hostFunctions, funcs...) + if closer != nil { + closers = append(closers, closer) + } + } + } + + // Check if the plugin needs to be recompiled with real host functions + compiled := info.compiled + needsRecompile := len(pluginManifest.AllowedHosts) > 0 || len(hostFunctions) > 0 + + // Recompile if needed. It is actually not a "recompile" since the first compilation + // should be cached by wazero. We just need to do it this way to provide the real host functions. + if needsRecompile { + log.Trace(m.ctx, "Recompiling plugin with host functions", "plugin", name) + info.compiled.Close(m.ctx) + extismConfig := extism.PluginConfig{ + EnableWasi: true, + RuntimeConfig: wazero.NewRuntimeConfig().WithCompilationCache(m.cache), + } + compiled, err = extism.NewCompiledPlugin(m.ctx, pluginManifest, extismConfig, hostFunctions) + if err != nil { + return err + } + } + + m.mu.Lock() + m.plugins[name] = &plugin{ + name: name, + path: wasmPath, + manifest: info.manifest, + compiled: compiled, + capabilities: capabilities, + closers: closers, + } + m.mu.Unlock() + + // Call plugin init function + callPluginInit(m.ctx, m.plugins[name]) + + return nil +} diff --git a/plugins/manager_plugin.go b/plugins/manager_plugin.go new file mode 100644 index 000000000..e719ee6a0 --- /dev/null +++ b/plugins/manager_plugin.go @@ -0,0 +1,43 @@ +package plugins + +import ( + "context" + "crypto/rand" + "errors" + "io" + + extism "github.com/extism/go-sdk" + "github.com/tetratelabs/wazero" +) + +// plugin represents a loaded plugin +type plugin struct { + name string // Plugin name (from filename) + path string // Path to the wasm file + manifest *Manifest + compiled *extism.CompiledPlugin + capabilities []Capability // Auto-detected capabilities based on exported functions + closers []io.Closer // Cleanup functions to call on unload +} + +func (p *plugin) instance() (*extism.Plugin, error) { + instance, err := p.compiled.Instance(context.Background(), extism.PluginInstanceConfig{ + ModuleConfig: wazero.NewModuleConfig().WithSysWalltime().WithRandSource(rand.Reader), + }) + if err != nil { + return nil, err + } + instance.SetLogger(extismLogger(p.name)) + return instance, nil +} + +func (p *plugin) Close() error { + var errs []error + for _, f := range p.closers { + err := f.Close() + if err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} diff --git a/plugins/manager_sync.go b/plugins/manager_sync.go new file mode 100644 index 000000000..8bb497772 --- /dev/null +++ b/plugins/manager_sync.go @@ -0,0 +1,226 @@ +package plugins + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + "github.com/navidrome/navidrome/log" + "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/model/request" +) + +// PluginMetadata holds the extracted information from a plugin file +// without fully initializing the plugin. +type PluginMetadata struct { + Manifest *Manifest + SHA256 string +} + +// adminContext returns a context with admin privileges for DB operations. +func adminContext(ctx context.Context) context.Context { + return request.WithUser(ctx, model.User{IsAdmin: true}) +} + +// marshalManifest marshals a manifest to JSON string, returning empty string on error. +func marshalManifest(m *Manifest) string { + b, _ := json.Marshal(m) + return string(b) +} + +// computeFileSHA256 computes the SHA-256 hash of a file without loading it into memory. +// This is used for quick change detection before full plugin compilation. +func computeFileSHA256(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +// addPluginToDB adds a new plugin to the database as disabled. +func (m *Manager) addPluginToDB(ctx context.Context, repo model.PluginRepository, name, path string, metadata *PluginMetadata) error { + now := time.Now() + newPlugin := &model.Plugin{ + ID: name, + Path: path, + Manifest: marshalManifest(metadata.Manifest), + SHA256: metadata.SHA256, + Enabled: false, + CreatedAt: now, + UpdatedAt: now, + } + if err := repo.Put(newPlugin); err != nil { + return fmt.Errorf("adding plugin to DB: %w", err) + } + log.Info(ctx, "Discovered new plugin", "plugin", name) + return nil +} + +// updatePluginInDB updates an existing plugin in the database after a file change. +// If the plugin was enabled, it will be unloaded and disabled. +func (m *Manager) updatePluginInDB(ctx context.Context, repo model.PluginRepository, dbPlugin *model.Plugin, path string, metadata *PluginMetadata) error { + wasEnabled := dbPlugin.Enabled + if wasEnabled { + if err := m.unloadPlugin(dbPlugin.ID); err != nil { + log.Debug(ctx, "Plugin not loaded during change", "plugin", dbPlugin.ID, err) + } + } + dbPlugin.Path = path + dbPlugin.Manifest = marshalManifest(metadata.Manifest) + dbPlugin.SHA256 = metadata.SHA256 + dbPlugin.Enabled = false + dbPlugin.LastError = "" + dbPlugin.UpdatedAt = time.Now() + if err := repo.Put(dbPlugin); err != nil { + return fmt.Errorf("updating plugin in DB: %w", err) + } + log.Info(ctx, "Plugin file changed", "plugin", dbPlugin.ID, "wasEnabled", wasEnabled) + return nil +} + +// removePluginFromDB removes a plugin from the database. +// If the plugin was enabled, it will be unloaded first. +func (m *Manager) removePluginFromDB(ctx context.Context, repo model.PluginRepository, dbPlugin *model.Plugin) error { + if dbPlugin.Enabled { + if err := m.unloadPlugin(dbPlugin.ID); err != nil { + log.Debug(ctx, "Plugin not loaded during removal", "plugin", dbPlugin.ID, err) + } + } + if err := repo.Delete(dbPlugin.ID); err != nil { + return fmt.Errorf("deleting plugin from DB: %w", err) + } + log.Info(ctx, "Plugin removed", "plugin", dbPlugin.ID) + return nil +} + +// syncPlugins scans the plugins folder and synchronizes with the database. +// It handles new, changed, and removed plugins by comparing SHA-256 hashes. +// - New plugins are added to DB as disabled +// - Changed plugins are updated in DB and disabled if they were enabled +// - Removed plugins are deleted from DB (after unloading if enabled) +func (m *Manager) syncPlugins(ctx context.Context, folder string) error { + if m.ds == nil { + return fmt.Errorf("datastore not configured") + } + + adminCtx := adminContext(ctx) + + // Read current plugins from folder + entries, err := os.ReadDir(folder) + if err != nil { + if os.IsNotExist(err) { + log.Debug(ctx, "Plugins folder does not exist", "folder", folder) + return nil + } + return fmt.Errorf("reading plugins folder: %w", err) + } + + // Build map of files in folder + filesOnDisk := make(map[string]string) // name -> path + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".wasm") { + continue + } + name := strings.TrimSuffix(entry.Name(), ".wasm") + filesOnDisk[name] = filepath.Join(folder, entry.Name()) + } + + // Get all plugins from DB + repo := m.ds.Plugin(adminCtx) + dbPlugins, err := repo.GetAll() + if err != nil { + return fmt.Errorf("reading plugins from DB: %w", err) + } + pluginsInDB := make(map[string]*model.Plugin) + for i := range dbPlugins { + pluginsInDB[dbPlugins[i].ID] = &dbPlugins[i] + } + + now := time.Now() + + // Process files on disk + for name, path := range filesOnDisk { + dbPlugin, exists := pluginsInDB[name] + + // Compute SHA256 first (lightweight operation) to check if plugin changed + sha256Hash, err := computeFileSHA256(path) + if err != nil { + log.Error(ctx, "Failed to compute SHA256 for plugin", "plugin", name, "path", path, err) + continue + } + + // If plugin exists in DB with same hash, skip full manifest extraction + if exists && dbPlugin.SHA256 == sha256Hash { + // Plugin unchanged - just update path in case folder moved + if dbPlugin.Path != path { + dbPlugin.Path = path + dbPlugin.UpdatedAt = now + if err := repo.Put(dbPlugin); err != nil { + log.Error(ctx, "Failed to update plugin path in DB", "plugin", name, err) + } + } + delete(pluginsInDB, name) + continue + } + + // Plugin is new or changed - need full manifest extraction + metadata, err := m.extractManifest(path) + if err != nil { + log.Error(ctx, "Failed to extract manifest from plugin", "plugin", name, "path", path, err) + // Store error in DB if plugin exists + if exists { + dbPlugin.LastError = err.Error() + dbPlugin.UpdatedAt = now + if dbPlugin.Enabled { + // Unload broken plugin + if unloadErr := m.unloadPlugin(name); unloadErr != nil { + log.Debug(ctx, "Plugin not loaded", "plugin", name) + } + dbPlugin.Enabled = false + } + if putErr := repo.Put(dbPlugin); putErr != nil { + log.Error(ctx, "Failed to update plugin in DB", "plugin", name, err) + } + } + delete(pluginsInDB, name) + continue + } + + if !exists { + // New plugin - add to DB as disabled + if err := m.addPluginToDB(ctx, repo, name, path, metadata); err != nil { + log.Error(ctx, "Failed to add plugin to DB", "plugin", name, err) + } + } else { + // Plugin changed - update DB + if err := m.updatePluginInDB(ctx, repo, dbPlugin, path, metadata); err != nil { + log.Error(ctx, "Failed to update plugin in DB", "plugin", name, err) + } + } + // Mark as processed + delete(pluginsInDB, name) + } + + // Remove plugins no longer on disk + for _, dbPlugin := range pluginsInDB { + if err := m.removePluginFromDB(ctx, repo, dbPlugin); err != nil { + log.Error(ctx, "Failed to delete plugin from DB", "plugin", dbPlugin.ID, err) + } + } + + return nil +} diff --git a/plugins/manager_test.go b/plugins/manager_test.go index 03c0da8b5..1ae10e1e7 100644 --- a/plugins/manager_test.go +++ b/plugins/manager_test.go @@ -3,12 +3,8 @@ package plugins import ( "context" "fmt" - "os" - "path/filepath" "sync" - "time" - "github.com/dustin/go-humanize" "github.com/navidrome/navidrome/core/agents" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -29,10 +25,10 @@ var _ = Describe("Manager", Ordered, func() { }) }) - Describe("UnloadPlugin", func() { + Describe("unloadPlugin", func() { It("removes a loaded plugin", func() { // Plugin is already loaded from Start - err := testManager.UnloadPlugin("test-metadata-agent") + err := testManager.unloadPlugin("test-metadata-agent") Expect(err).ToNot(HaveOccurred()) names := testManager.PluginNames(string(CapabilityMetadataAgent)) @@ -40,7 +36,7 @@ var _ = Describe("Manager", Ordered, func() { }) It("returns error when plugin not found", func() { - err := testManager.UnloadPlugin("nonexistent") + err := testManager.unloadPlugin("nonexistent") Expect(err).To(HaveOccurred()) Expect(err.Error()).To(ContainSubstring("not found")) }) @@ -126,178 +122,3 @@ var _ = Describe("Manager", Ordered, func() { } }) }) - -var _ = Describe("purgeCacheBySize", func() { - var ( - tmpDir string - ctx context.Context - ) - - BeforeEach(func() { - var err error - ctx = GinkgoT().Context() - tmpDir, err = os.MkdirTemp("", "cache-purge-test-*") - Expect(err).ToNot(HaveOccurred()) - }) - - AfterEach(func() { - os.RemoveAll(tmpDir) - }) - - createFileWithSize := func(path string, sizeBytes int64, modTime time.Time) { - dir := filepath.Dir(path) - err := os.MkdirAll(dir, 0755) - Expect(err).ToNot(HaveOccurred()) - - f, err := os.Create(path) - Expect(err).ToNot(HaveOccurred()) - defer f.Close() - - // Write random data to reach desired size - if sizeBytes > 0 { - err = f.Truncate(sizeBytes) - Expect(err).ToNot(HaveOccurred()) - } - - // Set modification time - err = os.Chtimes(path, modTime, modTime) - Expect(err).ToNot(HaveOccurred()) - } - - getDirSize := func(dir string) uint64 { - var total uint64 - err := filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error { - if err != nil || d.IsDir() { - return nil - } - info, err := d.Info() - if err != nil { - return nil - } - total += uint64(info.Size()) - return nil - }) - Expect(err).ToNot(HaveOccurred()) - return total - } - - Context("when maxSize is invalid or zero", func() { - It("should not remove any files with invalid size", func() { - cacheDir := filepath.Join(tmpDir, "cache") - createFileWithSize(filepath.Join(cacheDir, "file1.bin"), 1000, time.Now()) - createFileWithSize(filepath.Join(cacheDir, "file2.bin"), 1000, time.Now()) - - purgeCacheBySize(ctx, cacheDir, "invalid") - - Expect(getDirSize(cacheDir)).To(Equal(uint64(2000))) - }) - - It("should not remove any files when maxSize is 0", func() { - cacheDir := filepath.Join(tmpDir, "cache") - createFileWithSize(filepath.Join(cacheDir, "file1.bin"), 1000, time.Now()) - createFileWithSize(filepath.Join(cacheDir, "file2.bin"), 1000, time.Now()) - - purgeCacheBySize(ctx, cacheDir, "0") - - Expect(getDirSize(cacheDir)).To(Equal(uint64(2000))) - }) - }) - - Context("when cache directory doesn't exist", func() { - It("should not error", func() { - nonExistentDir := filepath.Join(tmpDir, "nonexistent") - Expect(func() { - purgeCacheBySize(ctx, nonExistentDir, "100MB") - }).ToNot(Panic()) - }) - }) - - Context("when total size is under limit", func() { - It("should not remove any files", func() { - cacheDir := filepath.Join(tmpDir, "cache") - createFileWithSize(filepath.Join(cacheDir, "file1.bin"), 1000, time.Now()) - createFileWithSize(filepath.Join(cacheDir, "file2.bin"), 1000, time.Now()) - - purgeCacheBySize(ctx, cacheDir, "10KB") - - Expect(getDirSize(cacheDir)).To(Equal(uint64(2000))) - }) - }) - - Context("when total size exceeds limit", func() { - It("should remove oldest files first", func() { - cacheDir := filepath.Join(tmpDir, "cache") - now := time.Now() - - // Create files with different ages (1MB each) - oldestFile := filepath.Join(cacheDir, "old.bin") - middleFile := filepath.Join(cacheDir, "middle.bin") - newestFile := filepath.Join(cacheDir, "new.bin") - - createFileWithSize(oldestFile, 1*1024*1024, now.Add(-3*time.Hour)) - createFileWithSize(middleFile, 1*1024*1024, now.Add(-2*time.Hour)) - createFileWithSize(newestFile, 1*1024*1024, now.Add(-1*time.Hour)) - - // Set limit to 2MiB - should remove oldest file - purgeCacheBySize(ctx, cacheDir, "2MiB") - - // Oldest should be removed - _, err := os.Stat(oldestFile) - Expect(os.IsNotExist(err)).To(BeTrue(), "oldest file should be removed") - - // Others should remain - _, err = os.Stat(middleFile) - Expect(err).ToNot(HaveOccurred(), "middle file should remain") - - _, err = os.Stat(newestFile) - Expect(err).ToNot(HaveOccurred(), "newest file should remain") - }) - - It("should remove multiple files to get under limit", func() { - cacheDir := filepath.Join(tmpDir, "cache") - now := time.Now() - - // Create 5 files, 1MiB each (total 5MiB) - for i := 0; i < 5; i++ { - path := filepath.Join(cacheDir, filepath.Join("dir", "file"+string(rune('0'+i))+".bin")) - createFileWithSize(path, 1*1024*1024, now.Add(-time.Duration(5-i)*time.Hour)) - } - - // Set limit to 2.5MiB - should remove oldest 3 files (leaving 2MiB) - purgeCacheBySize(ctx, cacheDir, "2.5MiB") - - finalSize := getDirSize(cacheDir) - limit, _ := humanize.ParseBytes("2.5MiB") - Expect(finalSize).To(BeNumerically("<=", limit)) - }) - - It("should remove empty parent directories after removing files", func() { - cacheDir := filepath.Join(tmpDir, "cache") - now := time.Now() - - // Create files in subdirectories - oldFile := filepath.Join(cacheDir, "subdir1", "old.bin") - newFile := filepath.Join(cacheDir, "subdir2", "new.bin") - - createFileWithSize(oldFile, 2*1024*1024, now.Add(-2*time.Hour)) - createFileWithSize(newFile, 2*1024*1024, now.Add(-1*time.Hour)) - - // Set limit to 2MiB - should remove old file and its parent dir - purgeCacheBySize(ctx, cacheDir, "2MiB") - - // Old file and its parent dir should be removed - _, err := os.Stat(oldFile) - Expect(os.IsNotExist(err)).To(BeTrue()) - - _, err = os.Stat(filepath.Join(cacheDir, "subdir1")) - Expect(os.IsNotExist(err)).To(BeTrue(), "empty parent directory should be removed") - - // New file and its parent dir should remain - _, err = os.Stat(newFile) - Expect(err).ToNot(HaveOccurred()) - - _, err = os.Stat(filepath.Join(cacheDir, "subdir2")) - Expect(err).ToNot(HaveOccurred()) - }) - }) -}) diff --git a/plugins/watcher.go b/plugins/manager_watcher.go similarity index 97% rename from plugins/watcher.go rename to plugins/manager_watcher.go index 58eb673b3..fb11e96bb 100644 --- a/plugins/watcher.go +++ b/plugins/manager_watcher.go @@ -154,7 +154,7 @@ func (m *Manager) processPluginEvent(pluginName string, eventType notify.Event) switch action { case actionAdd: // New file - extract manifest and add to DB as disabled - metadata, err := m.ExtractManifest(wasmPath) + metadata, err := m.extractManifest(wasmPath) if err != nil { log.Error(m.ctx, "Failed to extract manifest from new plugin", "plugin", pluginName, err) return @@ -174,7 +174,7 @@ func (m *Manager) processPluginEvent(pluginName string, eventType notify.Event) dbPlugin, err := repo.Get(pluginName) if err != nil { // Plugin not in DB yet, need full manifest extraction to add it - metadata, extractErr := m.ExtractManifest(wasmPath) + metadata, extractErr := m.extractManifest(wasmPath) if extractErr != nil { log.Error(m.ctx, "Failed to extract manifest from new plugin", "plugin", pluginName, extractErr) return @@ -191,14 +191,14 @@ func (m *Manager) processPluginEvent(pluginName string, eventType notify.Event) } // Plugin changed - now extract full manifest - metadata, err := m.ExtractManifest(wasmPath) + metadata, err := m.extractManifest(wasmPath) if err != nil { log.Error(m.ctx, "Failed to extract manifest from changed plugin", "plugin", pluginName, err) // Update error in DB dbPlugin.LastError = err.Error() dbPlugin.UpdatedAt = time.Now() if dbPlugin.Enabled { - _ = m.UnloadPlugin(pluginName) + _ = m.unloadPlugin(pluginName) dbPlugin.Enabled = false } _ = repo.Put(dbPlugin) diff --git a/plugins/watcher_test.go b/plugins/manager_watcher_test.go similarity index 98% rename from plugins/watcher_test.go rename to plugins/manager_watcher_test.go index 38205a3c7..39009dc66 100644 --- a/plugins/watcher_test.go +++ b/plugins/manager_watcher_test.go @@ -29,7 +29,7 @@ var _ = Describe("Plugin Watcher", func() { manager, tmpDir = createTestManager(nil) // Remove the auto-loaded plugin so tests can control loading - _ = manager.UnloadPlugin("test-metadata-agent") + _ = manager.unloadPlugin("test-metadata-agent") _ = os.Remove(filepath.Join(tmpDir, "test-metadata-agent.wasm")) }) @@ -48,7 +48,7 @@ var _ = Describe("Plugin Watcher", func() { AfterEach(func() { // Clean up: unload plugin if loaded, remove copied file - _ = manager.UnloadPlugin("test-metadata-agent") + _ = manager.unloadPlugin("test-metadata-agent") _ = os.Remove(filepath.Join(tmpDir, "test-metadata-agent.wasm")) })