package plugins import ( "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "log/slog" "os" "path/filepath" "runtime/debug" "slices" "strconv" "strings" "sync" "sync/atomic" "time" "golang.org/x/sync/singleflight" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/structpb" pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1" "github.com/Silo-Server/silo-server/internal/pluginhost" ) type pluginClient interface { Manifest() *pluginv1.PluginManifest MetadataProvider(capabilityID string) (*pluginhost.MetadataProviderClient, error) ImageResolver(capabilityID string) (*pluginhost.ImageResolverClient, error) MarkerProvider(capabilityID string) (*pluginhost.MarkerProviderClient, error) MediaAnalyzer(capabilityID string) (*pluginhost.MediaAnalyzerClient, error) ScheduledTask(capabilityID string) (*pluginhost.ScheduledTaskClient, error) ScanSource(capabilityID string) (*pluginhost.ScanSourceClient, error) RequestRouter(capabilityID string) (*pluginhost.RequestRouterClient, error) EventConsumer(capabilityID string) (*pluginhost.EventConsumerClient, error) AuthProvider(capabilityID string) (*pluginhost.AuthProviderClient, error) HTTPRoutes(capabilityID string) (*pluginhost.HTTPRoutesClient, error) WatchSyncProvider(capabilityID string) (*pluginhost.WatchSyncProviderClient, error) } type Host interface { Start(ctx context.Context, req pluginhost.StartRequest) (pluginClient, error) Client(installationID int) (pluginClient, error) Stop(installationID int) error Shutdown(ctx context.Context) error } type serviceInstallationStore interface { archiveStore GetByID(ctx context.Context, id int) (*Installation, error) List(ctx context.Context) ([]*Installation, error) ListEnabled(ctx context.Context) ([]*Installation, error) ListByPluginID(ctx context.Context, pluginID string) ([]*Installation, error) Update(ctx context.Context, id int, input UpdateInstallationInput) error ListCapabilities(ctx context.Context, installationID int) ([]*Capability, error) } type serviceConfigStore interface { ListGlobalConfigs(ctx context.Context, installationID int) ([]*RuntimeConfig, error) PutGlobalConfig(ctx context.Context, installationID int, key string, value map[string]any) error CompareAndSwapGlobalConfig( ctx context.Context, installationID int, key string, value map[string]any, expectedUpdatedAt *time.Time, ) (bool, error) } type Service struct { repositories *RepositoryStore installations serviceInstallationStore configs serviceConfigStore catalog *CatalogService installer *Installer archiveCache *ArchiveCache host Host testConfigSeq atomic.Int64 dispatcher *EventDispatcher lifecycleMu sync.RWMutex lifecycleHooks []func(context.Context) launchGroup singleflight.Group // installationCache memoizes plugin_installations rows keyed by ID so the // hot plugin-RPC path (ensureClient -> loadInstallation) and the metadata // chain enabled-check answer from memory instead of a per-call DB read. It // is wiped wholesale by invalidateInstallationCache, registered as a // lifecycle hook, so a cached row is at most one lifecycle event stale. // // installationCacheGen guards the read-through against an invalidate that // races an in-flight GetByID: the generation is captured before the store // read and re-checked under the write lock, so a row fetched before a // lifecycle mutation is never written back into a freshly-cleared cache. installationCacheMu sync.RWMutex installationCache map[int]*Installation installationCacheGen uint64 } // SetEventDispatcher wires the EventDispatcher into the Service. The // dispatcher reference is retained so future hooks can act on lifecycle // changes; the current dispatcher implementation is fully driven by // per-event store reads and needs no notification on install/enable/disable. func (s *Service) SetEventDispatcher(d *EventDispatcher) { s.dispatcher = d } // AddLifecycleHook registers a callback invoked after plugin install, enable, // disable, uninstall, preload, or runtime-configuration changes. func (s *Service) AddLifecycleHook(hook func(context.Context)) { if s == nil || hook == nil { return } s.lifecycleMu.Lock() s.lifecycleHooks = append(s.lifecycleHooks, hook) s.lifecycleMu.Unlock() } // OnLifecycleChange is invoked by API handlers and the installer after every // plugin lifecycle mutation. Hooks should be best-effort and log their own // errors so plugin admin operations are not failed by secondary cache refreshes. func (s *Service) OnLifecycleChange(ctx context.Context) { if s == nil { return } s.lifecycleMu.RLock() hooks := append([]func(context.Context){}, s.lifecycleHooks...) s.lifecycleMu.RUnlock() for _, hook := range hooks { func(hook func(context.Context)) { defer func() { if recovered := recover(); recovered != nil { slog.ErrorContext(ctx, "plugin lifecycle hook panicked; continuing", "component", "plugins", "panic", recovered, "stack", string(debug.Stack()), ) } }() hook(ctx) }(hook) } } func NewService( repositories *RepositoryStore, installations *InstallationStore, configs *RuntimeConfigStore, catalog *CatalogService, installer *Installer, host Host, ) *Service { svc := &Service{ repositories: repositories, installations: installations, configs: configs, catalog: catalog, installer: installer, archiveCache: NewArchiveCache(installations), host: host, } // Self-register the installation-cache invalidation so it can never be // silently forgotten by a new caller: every OnLifecycleChange (install / // enable / disable / update / uninstall) wipes the cache, keeping the // memoized rows correct without any external wiring. svc.AddLifecycleHook(func(context.Context) { svc.invalidateInstallationCache() }) return svc } func (s *Service) FetchCatalog(ctx context.Context) ([]CatalogEntry, error) { entries, err := s.catalog.Fetch(ctx) if err != nil { return nil, err } return catalogEntriesForDiscovery(entries), nil } func catalogEntriesForDiscovery(entries []CatalogEntry) []CatalogEntry { selected := make(map[string]CatalogEntry, len(entries)) for _, entry := range entries { if entry.Manifest == nil { continue } pluginID := entry.Manifest.GetPluginId() if isApprovedCommunityPlugin(pluginID) && entry.SourceKind != RepositorySourceApprovedCommunity { continue } existing, ok := selected[pluginID] if !ok || catalogEntryPreferredForDiscovery(entry, existing) { selected[pluginID] = entry } } result := make([]CatalogEntry, 0, len(selected)) for _, entry := range selected { result = append(result, entry) } slices.SortFunc(result, func(left, right CatalogEntry) int { return strings.Compare(left.Manifest.GetPluginId(), right.Manifest.GetPluginId()) }) return result } func catalogEntryPreferredForDiscovery(candidate, current CatalogEntry) bool { candidatePrecedence := repositorySourcePrecedence(candidate.SourceKind) currentPrecedence := repositorySourcePrecedence(current.SourceKind) if candidatePrecedence != currentPrecedence { return candidatePrecedence < currentPrecedence } if candidate.RepositoryID != current.RepositoryID { return candidate.RepositoryID < current.RepositoryID } return compareVersions(candidate.Manifest.GetVersion(), current.Manifest.GetVersion()) > 0 } func repositorySourcePrecedence(sourceKind string) int { switch sourceKind { case RepositorySourceSilo: return 0 case RepositorySourceApprovedCommunity: return 1 default: return 2 } } func (s *Service) InstallLocal(ctx context.Context, req InstallArchiveRequest) (*InstallResult, error) { if req.ArchivePath == "" { return nil, fmt.Errorf("archive path is required") } data, err := os.ReadFile(req.ArchivePath) if err != nil { return nil, fmt.Errorf("read archive %q: %w", req.ArchivePath, err) } _, _, manifest, err := openPluginArchive(data) if err != nil { return nil, err } existing, err := s.existingInstallationByPluginID(ctx, manifest.GetPluginId()) if err != nil { return nil, err } var result *InstallResult if existing != nil { if err := s.stopInstallationIfRunning(existing); err != nil { return nil, err } result, err = s.installer.ReplaceLocal(ctx, existing, req) } else { result, err = s.installer.InstallLocal(ctx, req) } if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } func (s *Service) InstallRemote(ctx context.Context, req InstallArchiveRequest) (*InstallResult, error) { result, err := s.installer.InstallRemote(ctx, req) if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } func (s *Service) InstallCatalog(ctx context.Context, req InstallCatalogRequest) (*InstallResult, error) { target, err := s.catalog.ResolveInstall(ctx, req) if err != nil { return nil, err } repositoryID := target.RepositoryID existing, err := s.existingInstallationByPluginID(ctx, req.PluginID) if err != nil { return nil, err } var result *InstallResult if target.LegacyArchive { archiveReq := InstallArchiveRequest{ ArchiveURL: target.ArchiveURL, RepositoryID: &repositoryID, } if existing == nil { result, err = s.installer.InstallRemote(ctx, archiveReq) } else { if err = s.stopInstallationIfRunning(existing); err != nil { return nil, err } result, err = s.installer.ReplaceRemote(ctx, existing, archiveReq) } } else { binaryReq := InstallBinaryRequest{ BinaryURL: target.ArchiveURL, Checksum: target.Checksum, RepositoryID: &repositoryID, } if existing == nil { result, err = s.installer.InstallBinary(ctx, binaryReq) } else { if err = s.stopInstallationIfRunning(existing); err != nil { return nil, err } result, err = s.installer.ReplaceBinary(ctx, existing, binaryReq) } } if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } // UpdateToAvailableVersion updates a plugin to its available_version. // Returns the updated installation after the update completes. func (s *Service) UpdateToAvailableVersion(ctx context.Context, installationID int) (*Installation, error) { installation, err := s.installations.GetByID(ctx, installationID) if err != nil { return nil, err } if installation.AvailableVersion == nil || *installation.AvailableVersion == "" { return nil, fmt.Errorf("no update available for plugin %q", installation.PluginID) } targetVersion := *installation.AvailableVersion if installation.RepositoryID == nil || *installation.RepositoryID == 0 { return nil, fmt.Errorf("plugin %q has no repository_id, cannot update from catalog", installation.PluginID) } _, err = s.InstallCatalog(ctx, InstallCatalogRequest{ RepositoryID: *installation.RepositoryID, PluginID: installation.PluginID, Version: targetVersion, }) if err != nil { return nil, fmt.Errorf("update plugin %q to %s: %w", installation.PluginID, targetVersion, err) } // Clear available_version now that we've updated. empty := "" if err := s.installations.Update(ctx, installationID, UpdateInstallationInput{ AvailableVersion: &empty, }); err != nil { slog.WarnContext(ctx, "failed to clear available_version after update", "component", "plugins", "installation_id", installationID, "error", err) } // Reload and return the updated installation. return s.installations.GetByID(ctx, installationID) } func (s *Service) InstallBinary(ctx context.Context, req InstallBinaryRequest) (*InstallResult, error) { var result *InstallResult var err error if req.Manifest != nil && req.Manifest.GetPluginId() != "" { existing, existErr := s.existingInstallationByPluginID(ctx, req.Manifest.GetPluginId()) if existErr != nil { return nil, existErr } if existing != nil { if err = s.stopInstallationIfRunning(existing); err != nil { return nil, err } result, err = s.installer.ReplaceBinary(ctx, existing, req) if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } } result, err = s.installer.InstallBinary(ctx, req) if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } func (s *Service) InstallBinaryUpload(ctx context.Context, binaryData []byte) (*InstallResult, error) { if len(binaryData) == 0 { return nil, fmt.Errorf("binary data is required") } manifest, err := loadManifestFromBinary(ctx, binaryData) if err != nil { return nil, err } checksum := sha256.Sum256(binaryData) actualChecksum := hex.EncodeToString(checksum[:]) var result *InstallResult if s.installations == nil { result, err = s.installer.installBinary(ctx, binaryData, actualChecksum, manifest, nil) if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } existing, err := s.installations.ListByPluginID(ctx, manifest.GetPluginId()) if err != nil { return nil, fmt.Errorf("list existing plugin installations for %q: %w", manifest.GetPluginId(), err) } if len(existing) == 0 { result, err = s.installer.installBinary(ctx, binaryData, actualChecksum, manifest, nil) if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } if len(existing) > 1 { return nil, fmt.Errorf("multiple existing installations found for plugin %q", manifest.GetPluginId()) } oldInstallation := existing[0] if err := s.stopInstallationIfRunning(oldInstallation); err != nil { return nil, err } result, err = s.installer.replaceBinary(ctx, oldInstallation, binaryData, actualChecksum, manifest) if err != nil { return nil, err } s.OnLifecycleChange(ctx) return result, nil } func (s *Service) existingInstallationByPluginID(ctx context.Context, pluginID string) (*Installation, error) { if s.installations == nil || pluginID == "" { return nil, nil } existing, err := s.installations.ListByPluginID(ctx, pluginID) if err != nil { return nil, fmt.Errorf("list existing plugin installations for %q: %w", pluginID, err) } if len(existing) == 0 { return nil, nil } if len(existing) > 1 { return nil, fmt.Errorf("multiple existing installations found for plugin %q", pluginID) } return existing[0], nil } func (s *Service) stopInstallationIfRunning(existing *Installation) error { if existing == nil || !existing.Enabled || s.host == nil { return nil } if err := s.host.Stop(existing.ID); err != nil && !errors.Is(err, pluginhost.ErrClientNotFound) { return fmt.Errorf("stop existing plugin installation %d: %w", existing.ID, err) } return nil } func (s *Service) PreloadEnabled(ctx context.Context) error { if s.installations == nil { return nil } installations, err := s.installations.ListEnabled(ctx) if err != nil { return err } for _, installation := range installations { if installation == nil { continue } // Builtin installations have no archive or binary; skip them explicitly // instead of leaning on the tolerated ErrArchiveNotFound branch below // (any other load error here is fatal to startup). if installation.IsBuiltin() { continue } if _, err := s.ensureLoadedInstallation(ctx, installation); err != nil { if errors.Is(err, ErrArchiveNotFound) { slog.WarnContext(ctx, "plugin preload skipped: archive not found for enabled installation", "component", "plugins", "installation_id", installation.ID, "plugin_id", installation.PluginID, "version", installation.Version, ) continue } return fmt.Errorf("preload plugin installation %d: %w", installation.ID, err) } } s.OnLifecycleChange(ctx) return nil } func (s *Service) Start(ctx context.Context, installationID int) (pluginClient, error) { installation, manifest, err := s.ensureInstallationCache(ctx, installationID, true) if err != nil { return nil, err } configEntries, err := s.globalConfigEntries(ctx, installation.ID) if err != nil { return nil, err } return s.host.Start(ctx, pluginhost.StartRequest{ InstallationID: installation.ID, BinaryPath: installation.InstallPath, Manifest: manifest, Config: configEntries, }) } func (s *Service) Stop(installationID int) error { if s.host == nil { return nil } return s.host.Stop(installationID) } func (s *Service) MediaAnalyzerClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.MediaAnalyzerClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.MediaAnalyzer(capabilityID) } func (s *Service) MetadataProviderClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.MetadataProviderClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.MetadataProvider(capabilityID) } func (s *Service) ImageResolverClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.ImageResolverClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.ImageResolver(capabilityID) } func (s *Service) MarkerProviderClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.MarkerProviderClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.MarkerProvider(capabilityID) } func (s *Service) ScheduledTaskClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.ScheduledTaskClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.ScheduledTask(capabilityID) } func (s *Service) ScanSourceClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.ScanSourceClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.ScanSource(capabilityID) } func (s *Service) RequestRouterClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.RequestRouterClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.RequestRouter(capabilityID) } func (s *Service) ScanSourceClientByPluginID( ctx context.Context, pluginID string, capabilityID string, ) (*pluginhost.ScanSourceClient, error) { if s == nil || s.installations == nil { return nil, fmt.Errorf("scan source plugin resolver is not configured") } installations, err := s.installations.ListByPluginID(ctx, pluginID) if err != nil { return nil, err } if len(installations) == 0 { return nil, fmt.Errorf("scan source plugin %q is not installed", pluginID) } if len(installations) > 1 { return nil, fmt.Errorf("scan source plugin %q is ambiguous across %d installations", pluginID, len(installations)) } var matches []*Installation for _, installation := range installations { if installation == nil { continue } capabilities, err := s.installations.ListCapabilities(ctx, installation.ID) if err != nil { return nil, err } for _, capability := range capabilities { if capability == nil { continue } if capability.Type == "scan_source.v1" && capability.ID == capabilityID { matches = append(matches, installation) break } } } if len(matches) == 0 { return nil, fmt.Errorf("scan source capability %q is not installed for plugin %q", capabilityID, pluginID) } if !matches[0].Enabled { return nil, fmt.Errorf("scan source plugin %q is disabled", pluginID) } return s.ScanSourceClient(ctx, matches[0].ID, capabilityID) } func (s *Service) WatchSyncProviderClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.WatchSyncProviderClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.WatchSyncProvider(capabilityID) } func (s *Service) EventConsumerClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.EventConsumerClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.EventConsumer(capabilityID) } func (s *Service) AuthProviderClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.AuthProviderClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.AuthProvider(capabilityID) } func (s *Service) HTTPRoutesClient( ctx context.Context, installationID int, capabilityID string, ) (*pluginhost.HTTPRoutesClient, error) { client, err := s.ensureClient(ctx, installationID) if err != nil { return nil, err } return client.HTTPRoutes(capabilityID) } func (s *Service) RouteDescriptors(ctx context.Context, installationID int) ([]*pluginv1.HttpRouteDescriptor, error) { manifest, err := s.manifestForInstallation(ctx, installationID, true) if err != nil { return nil, err } return append([]*pluginv1.HttpRouteDescriptor(nil), manifest.GetHttpRoutes()...), nil } func (s *Service) ResolveAssetPath(ctx context.Context, installationID int, assetPath string) (string, error) { installation, manifest, err := s.ensureInstallationCache(ctx, installationID, true) if err != nil { return "", err } for _, asset := range manifest.GetAssets() { if asset.GetPath() == assetPath { resolved := filepath.Join(filepath.Dir(installation.InstallPath), assetPath) if _, err := os.Stat(resolved); err != nil { return "", fmt.Errorf("plugin asset %q: %w", assetPath, err) } return resolved, nil } } return "", fmt.Errorf("plugin asset %q not found", assetPath) } func (s *Service) UserConfigSchema(ctx context.Context, installationID int) ([]*pluginv1.ConfigSchema, error) { manifest, err := s.manifestForInstallation(ctx, installationID, false) if err != nil { return nil, err } return append([]*pluginv1.ConfigSchema(nil), manifest.GetUserConfigSchema()...), nil } func (s *Service) ManifestForInstallation( ctx context.Context, installationID int, ) (*pluginv1.PluginManifest, error) { return s.manifestForInstallation(ctx, installationID, false) } // ensureClient returns a running client for the installation, collapsing // concurrent first-use of a cold installation into a single launch so a burst of // callers does not spawn redundant plugin processes (Host.Start releases its lock // during the slow launch and cannot dedupe). After the flight completes the key // is freed, so subsequent callers re-run and hit the now-warm cache. func (s *Service) ensureClient(ctx context.Context, installationID int) (pluginClient, error) { v, err, _ := s.launchGroup.Do(strconv.Itoa(installationID), func() (any, error) { // Isolate the shared launch from the leader caller's cancellation: other // waiters depend on this in-flight launch, so a single caller's canceled // request must not tear it down. Values (tracing, auth) are preserved. return s.doEnsureClient(context.WithoutCancel(ctx), installationID) }) if err != nil { return nil, err } return v.(pluginClient), nil } func (s *Service) doEnsureClient(ctx context.Context, installationID int) (pluginClient, error) { installation, err := s.loadInstallation(ctx, installationID, true) if err != nil { return nil, err } client, err := s.host.Client(installationID) if err == nil { installedManifest, manifestErr := LoadManifestFile(InstalledManifestPath(installation.InstallPath)) if manifestErr != nil { slog.WarnContext(ctx, "plugin installed manifest unavailable; reusing healthy client", "component", "plugins", "installation_id", installation.ID, "plugin_id", installation.PluginID, "version", installation.Version, "error", manifestErr, ) return client, nil } cachedManifest := client.Manifest() if cachedManifest != nil && proto.Equal(cachedManifest, installedManifest) { return client, nil } slog.WarnContext(ctx, "plugin client manifest drift detected; restarting", "component", "plugins", "installation_id", installation.ID, "plugin_id", installation.PluginID, "cached_plugin_id", manifestPluginID(cachedManifest), "installed_plugin_id", installedManifest.GetPluginId(), "cached_version", manifestVersion(cachedManifest), "installed_version", installedManifest.GetVersion(), ) if stopErr := s.host.Stop(installationID); stopErr != nil && !errors.Is(stopErr, pluginhost.ErrClientNotFound) { return nil, fmt.Errorf("stop stale plugin installation %d: %w", installationID, stopErr) } return s.Start(ctx, installationID) } if errors.Is(err, pluginhost.ErrPluginUnhealthy) { if stopErr := s.host.Stop(installationID); stopErr != nil && !errors.Is(stopErr, pluginhost.ErrClientNotFound) { return nil, fmt.Errorf("stop unhealthy plugin installation %d: %w", installationID, stopErr) } return s.Start(ctx, installationID) } if errors.Is(err, pluginhost.ErrClientNotFound) { return s.Start(ctx, installationID) } return nil, err } func (s *Service) manifestForInstallation(ctx context.Context, installationID int, requireEnabled bool) (*pluginv1.PluginManifest, error) { _, manifest, err := s.ensureInstallationCache(ctx, installationID, requireEnabled) return manifest, err } func (s *Service) ensureInstallationCache( ctx context.Context, installationID int, requireEnabled bool, ) (*Installation, *pluginv1.PluginManifest, error) { installation, err := s.loadInstallation(ctx, installationID, requireEnabled) if err != nil { return nil, nil, err } manifest, err := s.ensureLoadedInstallation(ctx, installation) if err != nil { return nil, nil, err } return installation, manifest, nil } func (s *Service) loadInstallation(ctx context.Context, installationID int, requireEnabled bool) (*Installation, error) { installation, err := s.cachedInstallation(ctx, installationID) if err != nil { return nil, err } // The requireEnabled gate is applied after the cache read so the cache // stores the row regardless of its enabled state and ErrInstallationDisabled // semantics are unchanged. if requireEnabled && !installation.Enabled { return nil, ErrInstallationDisabled } // requireEnabled marks paths that intend to launch or serve the plugin // (start, manifest routes/assets, gRPC clients, HTTP proxy). The reserved // builtin row has no binary behind it and must never reach those paths; // not-found gives the proxy and API a clean 4xx. Reads with // requireEnabled=false (IsInstallationEnabled for the metadata chain, // generic listings) still see the row. if requireEnabled && installation.IsBuiltin() { return nil, ErrInstallationNotFound } return installation, nil } // cachedInstallation returns the plugin_installations row for installationID // from the in-memory cache, loading it from the store on a miss. The returned // *Installation is shared and must be treated as read-only by callers; it is // evicted wholesale by invalidateInstallationCache on every lifecycle change. func (s *Service) cachedInstallation(ctx context.Context, installationID int) (*Installation, error) { s.installationCacheMu.RLock() cached, ok := s.installationCache[installationID] gen := s.installationCacheGen s.installationCacheMu.RUnlock() if ok { return cached, nil } installation, err := s.installations.GetByID(ctx, installationID) if err != nil { return nil, err } s.installationCacheMu.Lock() // Only publish the fetched row if no invalidation happened while GetByID was // in flight. Otherwise the row may pre-date a just-committed lifecycle change // (e.g. a disable), and writing it would resurrect stale state until the next // event. On a generation mismatch we still return the freshly-read row to the // caller but leave the cache untouched. if s.installationCacheGen == gen { if s.installationCache == nil { s.installationCache = make(map[int]*Installation) } s.installationCache[installationID] = installation } s.installationCacheMu.Unlock() return installation, nil } // invalidateInstallationCache clears the in-memory installation cache. It is // registered as a lifecycle hook (see NewService) so OnLifecycleChange evicts // stale rows after every install / enable / disable / update / uninstall. func (s *Service) invalidateInstallationCache() { if s == nil { return } s.installationCacheMu.Lock() s.installationCache = nil s.installationCacheGen++ s.installationCacheMu.Unlock() } // IsInstallationEnabled reports whether the given plugin installation is // enabled, served from the in-memory installation cache. It backs the metadata // chain's enabled-check (internal/metadata/chain.go) so provider construction // no longer issues a per-capability SELECT on the hot path. func (s *Service) IsInstallationEnabled(ctx context.Context, installationID int) (bool, error) { installation, err := s.loadInstallation(ctx, installationID, false) if err != nil { return false, err } return installation.Enabled, nil } // InstallationKind returns the installation's kind ("plugin" or "builtin") from // the same in-memory cache IsInstallationEnabled reads, so metadata chain // resolution can identify builtin rows without a per-capability DB query. func (s *Service) InstallationKind(ctx context.Context, installationID int) (string, error) { installation, err := s.loadInstallation(ctx, installationID, false) if err != nil { return "", err } return installation.Kind, nil } func (s *Service) ensureLoadedInstallation( ctx context.Context, installation *Installation, ) (*pluginv1.PluginManifest, error) { if s.archiveCache == nil { return LoadManifestFile(InstalledManifestPath(installation.InstallPath)) } return s.archiveCache.Ensure(ctx, installation) } func (s *Service) globalConfigEntries(ctx context.Context, installationID int) ([]*pluginv1.ConfigEntry, error) { if s.configs == nil { return nil, nil } configs, err := s.configs.ListGlobalConfigs(ctx, installationID) if err != nil { return nil, fmt.Errorf("list plugin runtime configs for installation %d: %w", installationID, err) } entries := make([]*pluginv1.ConfigEntry, 0, len(configs)) for _, config := range configs { if config == nil { continue } value := config.Value if value == nil { value = map[string]any{} } structValue, err := structpb.NewStruct(value) if err != nil { return nil, fmt.Errorf( "encode runtime config %q for installation %d: %w", config.Key, installationID, err, ) } entries = append(entries, &pluginv1.ConfigEntry{ Key: config.Key, Value: structValue, }) } return entries, nil }