Merge pull request #218 from Silo-Server/codex/image-resolver-capability
feat(plugins): add image resolver registry
This commit is contained in:
+180
-25
@@ -1074,33 +1074,14 @@ func main() {
|
||||
<-pluginAutoUpdateDone
|
||||
|
||||
imageResolver := metadata.NewPluginImageResolver()
|
||||
if pluginService != nil {
|
||||
// Register image resolver sources for all enabled plugin metadata providers.
|
||||
rows, err := deps.DB.Query(appCtx,
|
||||
`SELECT pc.plugin_installation_id, pc.capability_id
|
||||
FROM plugin_capabilities pc
|
||||
JOIN plugin_installations pi ON pi.id = pc.plugin_installation_id
|
||||
WHERE pc.capability_type = 'metadata_provider.v1' AND pi.enabled = true`)
|
||||
if err != nil {
|
||||
slog.Warn("failed to list capabilities for image resolver registration", "error", err)
|
||||
} else {
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var instID int
|
||||
var capID string
|
||||
if err := rows.Scan(&instID, &capID); err != nil {
|
||||
slog.Warn("failed to scan capability for image resolver", "error", err)
|
||||
continue
|
||||
}
|
||||
source := metadata.NewPluginClientSource(instID, capID, func(
|
||||
ctx context.Context, installationID int, capabilityID string,
|
||||
) (metadata.PluginMetadataClient, error) {
|
||||
return pluginService.MetadataProviderClient(ctx, installationID, capabilityID)
|
||||
})
|
||||
imageResolver.RegisterSource(capID, source)
|
||||
slog.Info("registered plugin image resolver", "capability_id", capID, "installation_id", instID)
|
||||
if pluginService != nil && installationStore != nil {
|
||||
reloadImageResolvers := func(ctx context.Context) {
|
||||
if err := reloadPluginImageResolvers(ctx, installationStore, imageResolver, pluginService); err != nil {
|
||||
slog.Warn("failed to reload plugin image resolvers", "error", err)
|
||||
}
|
||||
}
|
||||
pluginService.AddLifecycleHook(reloadImageResolvers)
|
||||
reloadImageResolvers(appCtx)
|
||||
}
|
||||
if deps.S3Public != nil {
|
||||
presignTTL := cfg.S3.MetadataPresignExpiry
|
||||
@@ -2484,6 +2465,180 @@ func configureS3Clients(cfg *config.Config, deps *api.Dependencies) {
|
||||
}
|
||||
}
|
||||
|
||||
type pluginImageResolverCapabilityStore interface {
|
||||
ListEnabled(ctx context.Context) ([]*plugins.Installation, error)
|
||||
ListCapabilities(ctx context.Context, installationID int) ([]*plugins.Capability, error)
|
||||
}
|
||||
|
||||
func reloadPluginImageResolvers(
|
||||
ctx context.Context,
|
||||
store pluginImageResolverCapabilityStore,
|
||||
resolver *metadata.PluginImageResolver,
|
||||
service *plugins.Service,
|
||||
) error {
|
||||
if resolver == nil {
|
||||
return nil
|
||||
}
|
||||
if store == nil || service == nil {
|
||||
resolver.ReplaceSources(nil)
|
||||
return nil
|
||||
}
|
||||
|
||||
installations, err := store.ListEnabled(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list enabled plugin installations: %w", err)
|
||||
}
|
||||
sort.Slice(installations, func(i, j int) bool {
|
||||
if installations[i] == nil {
|
||||
return false
|
||||
}
|
||||
if installations[j] == nil {
|
||||
return true
|
||||
}
|
||||
return installations[i].ID < installations[j].ID
|
||||
})
|
||||
|
||||
var registrations []metadata.PluginImageResolverSourceRegistration
|
||||
for _, installation := range installations {
|
||||
if installation == nil {
|
||||
continue
|
||||
}
|
||||
capabilities, err := store.ListCapabilities(ctx, installation.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list image resolver capabilities for installation %d: %w", installation.ID, err)
|
||||
}
|
||||
sort.Slice(capabilities, func(i, j int) bool {
|
||||
if capabilities[i] == nil {
|
||||
return false
|
||||
}
|
||||
if capabilities[j] == nil {
|
||||
return true
|
||||
}
|
||||
if capabilities[i].Type != capabilities[j].Type {
|
||||
return capabilities[i].Type < capabilities[j].Type
|
||||
}
|
||||
return capabilities[i].ID < capabilities[j].ID
|
||||
})
|
||||
|
||||
for _, capability := range capabilities {
|
||||
if capability == nil {
|
||||
continue
|
||||
}
|
||||
switch capability.Type {
|
||||
case sdkcapability.ImageResolver:
|
||||
schemes, priority := imageResolverCapabilityConfig(capability)
|
||||
if len(schemes) == 0 {
|
||||
slog.Warn("plugin image resolver capability has no valid schemes",
|
||||
"installation_id", installation.ID,
|
||||
"capability_id", capability.ID)
|
||||
continue
|
||||
}
|
||||
for _, scheme := range schemes {
|
||||
source := metadata.NewPluginClientSource(installation.ID, capability.ID, func(
|
||||
ctx context.Context, installationID int, capabilityID string,
|
||||
) (metadata.PluginMetadataClient, error) {
|
||||
return service.ImageResolverClient(ctx, installationID, capabilityID)
|
||||
})
|
||||
registrations = append(registrations, metadata.PluginImageResolverSourceRegistration{
|
||||
Scheme: scheme,
|
||||
Source: source,
|
||||
Kind: metadata.PluginImageResolverSourceExplicit,
|
||||
Priority: priority,
|
||||
InstallationID: installation.ID,
|
||||
CapabilityID: capability.ID,
|
||||
})
|
||||
}
|
||||
case sdkcapability.MetadataProvider:
|
||||
scheme := strings.TrimSpace(capability.ID)
|
||||
if !metadata.ValidImageResolverScheme(scheme) {
|
||||
slog.Warn("skipping legacy metadata image resolver with invalid scheme",
|
||||
"installation_id", installation.ID,
|
||||
"capability_id", capability.ID)
|
||||
continue
|
||||
}
|
||||
source := metadata.NewPluginClientSource(installation.ID, capability.ID, func(
|
||||
ctx context.Context, installationID int, capabilityID string,
|
||||
) (metadata.PluginMetadataClient, error) {
|
||||
return service.MetadataProviderClient(ctx, installationID, capabilityID)
|
||||
})
|
||||
registrations = append(registrations, metadata.PluginImageResolverSourceRegistration{
|
||||
Scheme: scheme,
|
||||
Source: source,
|
||||
Kind: metadata.PluginImageResolverSourceLegacy,
|
||||
InstallationID: installation.ID,
|
||||
CapabilityID: capability.ID,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
resolver.ReplaceSources(registrations)
|
||||
slog.Info("reloaded plugin image resolvers", "sources", len(registrations))
|
||||
return nil
|
||||
}
|
||||
|
||||
func imageResolverCapabilityConfig(capability *plugins.Capability) ([]string, int) {
|
||||
if capability == nil {
|
||||
return nil, 0
|
||||
}
|
||||
meta := capabilityMetadataFields(capability.Metadata)
|
||||
return metadataStringList(meta["schemes"]), metadataInt(meta["priority"])
|
||||
}
|
||||
|
||||
func capabilityMetadataFields(raw map[string]any) map[string]any {
|
||||
if raw == nil {
|
||||
return nil
|
||||
}
|
||||
if nested, ok := raw["metadata"]; ok {
|
||||
switch typed := nested.(type) {
|
||||
case map[string]any:
|
||||
return typed
|
||||
}
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
func metadataStringList(value any) []string {
|
||||
var out []string
|
||||
switch typed := value.(type) {
|
||||
case []string:
|
||||
for _, item := range typed {
|
||||
if scheme := strings.TrimSpace(item); metadata.ValidImageResolverScheme(scheme) {
|
||||
out = append(out, scheme)
|
||||
}
|
||||
}
|
||||
case []any:
|
||||
for _, item := range typed {
|
||||
text, ok := item.(string)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if scheme := strings.TrimSpace(text); metadata.ValidImageResolverScheme(scheme) {
|
||||
out = append(out, scheme)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func metadataInt(value any) int {
|
||||
switch typed := value.(type) {
|
||||
case int:
|
||||
return typed
|
||||
case int32:
|
||||
return int(typed)
|
||||
case int64:
|
||||
return int(typed)
|
||||
case float64:
|
||||
return int(typed)
|
||||
case json.Number:
|
||||
n, _ := typed.Int64()
|
||||
return int(n)
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
type markerPluginCapabilityStore interface {
|
||||
ListEnabled(ctx context.Context) ([]*plugins.Installation, error)
|
||||
ListCapabilities(ctx context.Context, installationID int) ([]*plugins.Capability, error)
|
||||
|
||||
@@ -77,7 +77,7 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/Silo-Server/silo-plugin-sdk v0.7.0
|
||||
github.com/Silo-Server/silo-plugin-sdk v0.8.1
|
||||
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 // indirect
|
||||
|
||||
@@ -4,8 +4,8 @@ github.com/PuerkitoBio/goquery v1.8.0 h1:PJTF7AmFCFKk1N6V6jmKfrNH9tV5pNE6lZMkG0g
|
||||
github.com/PuerkitoBio/goquery v1.8.0/go.mod h1:ypIiRMtY7COPGk+I/YbZLbxsxn9g5ejnI2HSMtkjZvI=
|
||||
github.com/SherClockHolmes/webpush-go v1.4.0 h1:ocnzNKWN23T9nvHi6IfyrQjkIc0oJWv1B1pULsf9i3s=
|
||||
github.com/SherClockHolmes/webpush-go v1.4.0/go.mod h1:XSq8pKX11vNV8MJEMwjrlTkxhAj1zKfxmyhdV7Pd6UA=
|
||||
github.com/Silo-Server/silo-plugin-sdk v0.7.0 h1:VbD7qXjwKYOdajFBYQq1eS2fgHDWyShZBYh0kxJly6U=
|
||||
github.com/Silo-Server/silo-plugin-sdk v0.7.0/go.mod h1:etqmxLTwjxpFH9goAjBDfNDoqHMv2/sqUXu8yx3hNfA=
|
||||
github.com/Silo-Server/silo-plugin-sdk v0.8.1 h1:6DLm2s/O3N9KTx2d09Nq3bbV8z4u9HJwO2gueQDPyPw=
|
||||
github.com/Silo-Server/silo-plugin-sdk v0.8.1/go.mod h1:etqmxLTwjxpFH9goAjBDfNDoqHMv2/sqUXu8yx3hNfA=
|
||||
github.com/abadojack/whatlanggo v1.0.1 h1:19N6YogDnf71CTHm3Mp2qhYfkRdyvbgwWdd2EPxJRG4=
|
||||
github.com/abadojack/whatlanggo v1.0.1/go.mod h1:66WiQbSbJBIlOZMsvbKe5m6pzQovxCH9B/K8tQB2uoc=
|
||||
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
|
||||
|
||||
@@ -503,7 +503,6 @@ func (h *PluginHandler) HandleCreateInstallation(w http.ResponseWriter, r *http.
|
||||
}
|
||||
|
||||
h.syncMetadataProviders(r.Context(), result.Installation)
|
||||
h.syncImageResolvers(r.Context(), result.Installation)
|
||||
|
||||
response, err := h.buildInstallationResponse(r.Context(), result.Installation, result.Manifest)
|
||||
if err != nil {
|
||||
@@ -662,7 +661,6 @@ func (h *PluginHandler) installUploadedPlugin(ctx context.Context, path string)
|
||||
|
||||
func (h *PluginHandler) writeUploadedPluginResponse(w http.ResponseWriter, r *http.Request, result *plugins.InstallResult) {
|
||||
h.syncMetadataProviders(r.Context(), result.Installation)
|
||||
h.syncImageResolvers(r.Context(), result.Installation)
|
||||
|
||||
response, err := h.buildInstallationResponse(r.Context(), result.Installation, result.Manifest)
|
||||
if err != nil {
|
||||
@@ -703,35 +701,6 @@ func (h *PluginHandler) syncMetadataProviders(ctx context.Context, installation
|
||||
}
|
||||
}
|
||||
|
||||
// syncImageResolvers registers image resolver sources for any metadata_provider.v1
|
||||
// capabilities from the given installation so that plugin-prefixed image URLs
|
||||
// (e.g. "tmdb://poster/abc.jpg") can be resolved at runtime.
|
||||
func (h *PluginHandler) syncImageResolvers(ctx context.Context, installation *plugins.Installation) {
|
||||
if h.imageResolver == nil || h.service == nil {
|
||||
return
|
||||
}
|
||||
|
||||
caps, err := h.installations.ListCapabilities(ctx, installation.ID)
|
||||
if err != nil {
|
||||
slog.Error("listing capabilities for image resolver sync",
|
||||
"installation_id", installation.ID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, cap := range caps {
|
||||
if cap.Type != "metadata_provider.v1" {
|
||||
continue
|
||||
}
|
||||
source := metadata.NewPluginClientSource(installation.ID, cap.ID, func(
|
||||
ctx context.Context, installationID int, capabilityID string,
|
||||
) (metadata.PluginMetadataClient, error) {
|
||||
return h.service.MetadataProviderClient(ctx, installationID, capabilityID)
|
||||
})
|
||||
h.imageResolver.RegisterSource(cap.ID, source)
|
||||
slog.Info("registered plugin image resolver", "capability_id", cap.ID, "installation_id", installation.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func isZipUpload(data []byte) bool {
|
||||
if len(data) < 4 {
|
||||
return false
|
||||
@@ -849,11 +818,6 @@ func (h *PluginHandler) HandleUpdateInstallation(w http.ResponseWriter, r *http.
|
||||
return
|
||||
}
|
||||
|
||||
// When a plugin is being enabled, register its image resolvers.
|
||||
if req.Enabled != nil && *req.Enabled && !currentInstallation.Enabled {
|
||||
h.syncImageResolvers(r.Context(), installation)
|
||||
}
|
||||
|
||||
response, err := h.buildInstallationResponse(r.Context(), installation, nil)
|
||||
if err != nil {
|
||||
slog.Error("building updated plugin installation response", "error", err)
|
||||
@@ -883,7 +847,6 @@ func (h *PluginHandler) HandleApplyUpdate(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
|
||||
h.syncMetadataProviders(r.Context(), installation)
|
||||
h.syncImageResolvers(r.Context(), installation)
|
||||
|
||||
response, err := h.buildInstallationResponse(r.Context(), installation, nil)
|
||||
if err != nil {
|
||||
|
||||
@@ -15,6 +15,8 @@ import (
|
||||
|
||||
pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
|
||||
"golang.org/x/sync/singleflight"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -33,12 +35,36 @@ type expiringPluginImageResolverSource interface {
|
||||
ResolveImageURLsWithExpiry(ctx context.Context, paths []string, variant string) (map[string]catalog.ResolvedImageURL, error)
|
||||
}
|
||||
|
||||
type PluginImageResolverSourceKind string
|
||||
|
||||
const (
|
||||
PluginImageResolverSourceExplicit PluginImageResolverSourceKind = "explicit"
|
||||
PluginImageResolverSourceLegacy PluginImageResolverSourceKind = "legacy"
|
||||
)
|
||||
|
||||
type PluginImageResolverSourceRegistration struct {
|
||||
Scheme string
|
||||
Source PluginImageResolverSource
|
||||
Kind PluginImageResolverSourceKind
|
||||
Priority int
|
||||
InstallationID int
|
||||
CapabilityID string
|
||||
}
|
||||
|
||||
type pluginImageResolverSourceEntry struct {
|
||||
source PluginImageResolverSource
|
||||
kind PluginImageResolverSourceKind
|
||||
priority int
|
||||
installationID int
|
||||
capabilityID string
|
||||
}
|
||||
|
||||
// PluginImageResolver resolves plugin-prefixed image paths (e.g., "metadb://images/abc/original.jpg")
|
||||
// by parsing the prefix, routing to the correct plugin, and returning resolved URLs.
|
||||
// It implements catalog.ImageResolver and the catalog expiry-aware resolver extension.
|
||||
type PluginImageResolver struct {
|
||||
mu sync.RWMutex
|
||||
sources map[string]PluginImageResolverSource
|
||||
sources map[string][]pluginImageResolverSourceEntry
|
||||
s3Presigner s3ImagePresigner
|
||||
s3PresignTTL time.Duration
|
||||
urlCache *cache.TTLCache[catalog.ResolvedImageURL]
|
||||
@@ -48,7 +74,7 @@ type PluginImageResolver struct {
|
||||
// NewPluginImageResolver creates a new resolver with no registered sources.
|
||||
func NewPluginImageResolver() *PluginImageResolver {
|
||||
return &PluginImageResolver{
|
||||
sources: make(map[string]PluginImageResolverSource),
|
||||
sources: make(map[string][]pluginImageResolverSourceEntry),
|
||||
s3PresignTTL: 15 * time.Minute,
|
||||
urlCache: cache.NewTTLCache[catalog.ResolvedImageURL](),
|
||||
}
|
||||
@@ -62,9 +88,53 @@ type s3ImagePresigner interface {
|
||||
// RegisterSource registers a plugin provider as a source for resolving images
|
||||
// with the given plugin ID prefix.
|
||||
func (r *PluginImageResolver) RegisterSource(pluginID string, source PluginImageResolverSource) {
|
||||
if source == nil || !ValidImageResolverScheme(pluginID) {
|
||||
return
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.sources[pluginID] = source
|
||||
r.sources[pluginID] = append(r.sources[pluginID], pluginImageResolverSourceEntry{
|
||||
source: source,
|
||||
kind: PluginImageResolverSourceLegacy,
|
||||
})
|
||||
sortImageResolverSources(r.sources[pluginID])
|
||||
r.urlCache.InvalidatePrefix("")
|
||||
}
|
||||
|
||||
func (r *PluginImageResolver) ReplaceSources(registrations []PluginImageResolverSourceRegistration) {
|
||||
sources := make(map[string][]pluginImageResolverSourceEntry)
|
||||
for _, registration := range registrations {
|
||||
scheme := strings.TrimSpace(registration.Scheme)
|
||||
if registration.Source == nil || !ValidImageResolverScheme(scheme) {
|
||||
continue
|
||||
}
|
||||
kind := registration.Kind
|
||||
if kind == "" {
|
||||
kind = PluginImageResolverSourceLegacy
|
||||
}
|
||||
sources[scheme] = append(sources[scheme], pluginImageResolverSourceEntry{
|
||||
source: registration.Source,
|
||||
kind: kind,
|
||||
priority: registration.Priority,
|
||||
installationID: registration.InstallationID,
|
||||
capabilityID: registration.CapabilityID,
|
||||
})
|
||||
}
|
||||
for scheme := range sources {
|
||||
sortImageResolverSources(sources[scheme])
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
r.sources = sources
|
||||
r.mu.Unlock()
|
||||
r.urlCache.InvalidatePrefix("")
|
||||
}
|
||||
|
||||
func ValidImageResolverScheme(scheme string) bool {
|
||||
return scheme != "" &&
|
||||
scheme == strings.TrimSpace(scheme) &&
|
||||
scheme == strings.ToLower(scheme) &&
|
||||
!strings.Contains(scheme, "://")
|
||||
}
|
||||
|
||||
func (r *PluginImageResolver) SetS3Presigner(presigner s3ImagePresigner, ttl time.Duration) {
|
||||
@@ -143,13 +213,13 @@ func (r *PluginImageResolver) ResolveImageURLsWithExpiry(ctx context.Context, pa
|
||||
r.mu.RLock()
|
||||
presigner := r.s3Presigner
|
||||
s3TTL := r.s3PresignTTL
|
||||
sourcesSnapshot := make(map[string]PluginImageResolverSource, len(grouped))
|
||||
sourcesSnapshot := make(map[string][]pluginImageResolverSourceEntry, len(grouped))
|
||||
for pluginID := range grouped {
|
||||
if pluginID == "" {
|
||||
continue
|
||||
}
|
||||
if source, ok := r.sources[pluginID]; ok {
|
||||
sourcesSnapshot[pluginID] = source
|
||||
if sources, ok := r.sources[pluginID]; ok {
|
||||
sourcesSnapshot[pluginID] = append([]pluginImageResolverSourceEntry(nil), sources...)
|
||||
}
|
||||
}
|
||||
r.mu.RUnlock()
|
||||
@@ -161,12 +231,12 @@ func (r *PluginImageResolver) ResolveImageURLsWithExpiry(ctx context.Context, pa
|
||||
if pluginID == "" {
|
||||
return r.resolveS3Batch(ctx, presigner, s3TTL, entries), nil
|
||||
}
|
||||
source, ok := sourcesSnapshot[pluginID]
|
||||
if !ok {
|
||||
slog.Warn("no image resolver registered for plugin", "plugin_id", pluginID)
|
||||
sources := sourcesSnapshot[pluginID]
|
||||
if len(sources) == 0 {
|
||||
slog.Warn("no image resolver registered for scheme", "scheme", pluginID)
|
||||
return map[string]catalog.ResolvedImageURL{}, nil
|
||||
}
|
||||
return r.resolvePluginBatch(ctx, pluginID, source, entries, variant), nil
|
||||
return r.resolvePluginBatchWithFallback(ctx, pluginID, sources, entries, variant), nil
|
||||
})
|
||||
if err != nil {
|
||||
slog.Error("image batch resolution failed", "plugin_id", pluginID, "error", err)
|
||||
@@ -189,6 +259,53 @@ func (r *PluginImageResolver) ResolveImageURLsWithExpiry(ctx context.Context, pa
|
||||
return result
|
||||
}
|
||||
|
||||
func (r *PluginImageResolver) resolvePluginBatchWithFallback(
|
||||
ctx context.Context,
|
||||
pluginID string,
|
||||
sources []pluginImageResolverSourceEntry,
|
||||
entries []resolveEntry,
|
||||
variant string,
|
||||
) map[string]catalog.ResolvedImageURL {
|
||||
resolved := make(map[string]catalog.ResolvedImageURL, len(entries))
|
||||
remaining := append([]resolveEntry(nil), entries...)
|
||||
|
||||
for _, source := range sources {
|
||||
if len(remaining) == 0 {
|
||||
break
|
||||
}
|
||||
resolvedBatch, err := r.resolvePluginBatch(ctx, source.source, remaining, variant)
|
||||
if err != nil {
|
||||
if status.Code(err) == codes.Unimplemented {
|
||||
slog.Debug("plugin image resolver source does not implement image resolution",
|
||||
"scheme", pluginID,
|
||||
"source_kind", source.kind,
|
||||
"installation_id", source.installationID,
|
||||
"capability_id", source.capabilityID)
|
||||
continue
|
||||
}
|
||||
slog.Error("plugin batch image resolution failed",
|
||||
"scheme", pluginID,
|
||||
"source_kind", source.kind,
|
||||
"installation_id", source.installationID,
|
||||
"capability_id", source.capabilityID,
|
||||
"error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
nextRemaining := remaining[:0]
|
||||
for _, entry := range remaining {
|
||||
if value, ok := resolvedBatch[entry.originalPath]; ok && value.URL != "" {
|
||||
resolved[entry.originalPath] = value
|
||||
continue
|
||||
}
|
||||
nextRemaining = append(nextRemaining, entry)
|
||||
}
|
||||
remaining = nextRemaining
|
||||
}
|
||||
|
||||
return resolved
|
||||
}
|
||||
|
||||
func (r *PluginImageResolver) resolveS3Batch(
|
||||
ctx context.Context,
|
||||
presigner s3ImagePresigner,
|
||||
@@ -214,11 +331,10 @@ func (r *PluginImageResolver) resolveS3Batch(
|
||||
|
||||
func (r *PluginImageResolver) resolvePluginBatch(
|
||||
ctx context.Context,
|
||||
pluginID string,
|
||||
source PluginImageResolverSource,
|
||||
entries []resolveEntry,
|
||||
variant string,
|
||||
) map[string]catalog.ResolvedImageURL {
|
||||
) (map[string]catalog.ResolvedImageURL, error) {
|
||||
barePaths := make([]string, len(entries))
|
||||
for i, entry := range entries {
|
||||
barePaths[i] = entry.barePath
|
||||
@@ -239,8 +355,7 @@ func (r *PluginImageResolver) resolvePluginBatch(
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
slog.Error("plugin batch image resolution failed", "plugin_id", pluginID, "error", err)
|
||||
return map[string]catalog.ResolvedImageURL{}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resolved := make(map[string]catalog.ResolvedImageURL, len(entries))
|
||||
@@ -249,7 +364,7 @@ func (r *PluginImageResolver) resolvePluginBatch(
|
||||
resolved[entry.originalPath] = value
|
||||
}
|
||||
}
|
||||
return resolved
|
||||
return resolved, nil
|
||||
}
|
||||
|
||||
type resolveEntry struct {
|
||||
@@ -268,6 +383,28 @@ func sortedResolveEntries(entriesByOriginal map[string]resolveEntry) []resolveEn
|
||||
return entries
|
||||
}
|
||||
|
||||
func sortImageResolverSources(sources []pluginImageResolverSourceEntry) {
|
||||
sort.SliceStable(sources, func(i, j int) bool {
|
||||
if sourceKindRank(sources[i].kind) != sourceKindRank(sources[j].kind) {
|
||||
return sourceKindRank(sources[i].kind) < sourceKindRank(sources[j].kind)
|
||||
}
|
||||
if sources[i].priority != sources[j].priority {
|
||||
return sources[i].priority > sources[j].priority
|
||||
}
|
||||
if sources[i].installationID != sources[j].installationID {
|
||||
return sources[i].installationID < sources[j].installationID
|
||||
}
|
||||
return sources[i].capabilityID < sources[j].capabilityID
|
||||
})
|
||||
}
|
||||
|
||||
func sourceKindRank(kind PluginImageResolverSourceKind) int {
|
||||
if kind == PluginImageResolverSourceExplicit {
|
||||
return 0
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
func resolvedImageCacheKey(variant, path string) string {
|
||||
return variant + "\x00" + path
|
||||
}
|
||||
|
||||
@@ -9,6 +9,8 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/Silo-Server/silo-server/internal/catalog"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
type fakeExpiringImageSource struct {
|
||||
@@ -61,6 +63,46 @@ func (s *fakeExpiringImageSource) ResolveImageURLsWithExpiry(ctx context.Context
|
||||
return resolved, nil
|
||||
}
|
||||
|
||||
type scriptedImageSource struct {
|
||||
urls map[string]string
|
||||
err error
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (s *scriptedImageSource) ResolveImageURL(ctx context.Context, path string, variant string) (string, error) {
|
||||
resolved, err := s.ResolveImageURLsWithExpiry(ctx, []string{path}, variant)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return resolved[path].URL, nil
|
||||
}
|
||||
|
||||
func (s *scriptedImageSource) ResolveImageURLs(ctx context.Context, paths []string, variant string) (map[string]string, error) {
|
||||
resolved, err := s.ResolveImageURLsWithExpiry(ctx, paths, variant)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
urls := make(map[string]string, len(resolved))
|
||||
for path, value := range resolved {
|
||||
urls[path] = value.URL
|
||||
}
|
||||
return urls, nil
|
||||
}
|
||||
|
||||
func (s *scriptedImageSource) ResolveImageURLsWithExpiry(_ context.Context, paths []string, variant string) (map[string]catalog.ResolvedImageURL, error) {
|
||||
s.calls.Add(1)
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
}
|
||||
resolved := make(map[string]catalog.ResolvedImageURL, len(paths))
|
||||
for _, path := range paths {
|
||||
if url, ok := s.urls[path]; ok {
|
||||
resolved[path] = catalog.ResolvedImageURL{URL: url + ":" + variant}
|
||||
}
|
||||
}
|
||||
return resolved, nil
|
||||
}
|
||||
|
||||
func TestPluginImageResolverCachesOnlyKnownUsableExpiries(t *testing.T) {
|
||||
expiresAt := time.Now().Add(time.Hour)
|
||||
source := &fakeExpiringImageSource{expiresAt: &expiresAt}
|
||||
@@ -102,6 +144,159 @@ func TestPluginImageResolverCachesOnlyKnownUsableExpiries(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginImageResolverExplicitSourcesPrecedeLegacyFallbacks(t *testing.T) {
|
||||
resolver := NewPluginImageResolver()
|
||||
defer resolver.Close()
|
||||
|
||||
explicit := &scriptedImageSource{urls: map[string]string{"poster.jpg": "explicit"}}
|
||||
legacy := &scriptedImageSource{urls: map[string]string{"poster.jpg": "legacy"}}
|
||||
resolver.ReplaceSources([]PluginImageResolverSourceRegistration{
|
||||
{
|
||||
Scheme: "tmdb",
|
||||
Source: legacy,
|
||||
Kind: PluginImageResolverSourceLegacy,
|
||||
Priority: 1000,
|
||||
InstallationID: 1,
|
||||
CapabilityID: "tmdb",
|
||||
},
|
||||
{
|
||||
Scheme: "tmdb",
|
||||
Source: explicit,
|
||||
Kind: PluginImageResolverSourceExplicit,
|
||||
Priority: 0,
|
||||
InstallationID: 2,
|
||||
CapabilityID: "tmdb",
|
||||
},
|
||||
})
|
||||
|
||||
got := resolver.ResolveImageURL(context.Background(), "tmdb://poster.jpg", "card")
|
||||
if got != "explicit:card" {
|
||||
t.Fatalf("resolved URL = %q, want explicit source", got)
|
||||
}
|
||||
if calls := legacy.calls.Load(); calls != 0 {
|
||||
t.Fatalf("legacy source calls = %d, want 0 when explicit resolves", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginImageResolverOrdersSourcesByPriority(t *testing.T) {
|
||||
resolver := NewPluginImageResolver()
|
||||
defer resolver.Close()
|
||||
|
||||
low := &scriptedImageSource{urls: map[string]string{"poster.jpg": "low"}}
|
||||
high := &scriptedImageSource{urls: map[string]string{"poster.jpg": "high"}}
|
||||
resolver.ReplaceSources([]PluginImageResolverSourceRegistration{
|
||||
{
|
||||
Scheme: "tmdb",
|
||||
Source: low,
|
||||
Kind: PluginImageResolverSourceExplicit,
|
||||
Priority: 10,
|
||||
InstallationID: 1,
|
||||
CapabilityID: "low",
|
||||
},
|
||||
{
|
||||
Scheme: "tmdb",
|
||||
Source: high,
|
||||
Kind: PluginImageResolverSourceExplicit,
|
||||
Priority: 50,
|
||||
InstallationID: 2,
|
||||
CapabilityID: "high",
|
||||
},
|
||||
})
|
||||
|
||||
got := resolver.ResolveImageURL(context.Background(), "tmdb://poster.jpg", "card")
|
||||
if got != "high:card" {
|
||||
t.Fatalf("resolved URL = %q, want high-priority source", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginImageResolverSkipsUnimplementedSourcesInEitherRegistrationOrder(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
registrations func(broken, working *scriptedImageSource) []PluginImageResolverSourceRegistration
|
||||
}{
|
||||
{
|
||||
name: "working registered first",
|
||||
registrations: func(broken, working *scriptedImageSource) []PluginImageResolverSourceRegistration {
|
||||
return []PluginImageResolverSourceRegistration{
|
||||
{Scheme: "tmdb", Source: working, Kind: PluginImageResolverSourceExplicit, Priority: 10, InstallationID: 2, CapabilityID: "working"},
|
||||
{Scheme: "tmdb", Source: broken, Kind: PluginImageResolverSourceExplicit, Priority: 20, InstallationID: 1, CapabilityID: "broken"},
|
||||
}
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "broken registered first",
|
||||
registrations: func(broken, working *scriptedImageSource) []PluginImageResolverSourceRegistration {
|
||||
return []PluginImageResolverSourceRegistration{
|
||||
{Scheme: "tmdb", Source: broken, Kind: PluginImageResolverSourceExplicit, Priority: 20, InstallationID: 1, CapabilityID: "broken"},
|
||||
{Scheme: "tmdb", Source: working, Kind: PluginImageResolverSourceExplicit, Priority: 10, InstallationID: 2, CapabilityID: "working"},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
resolver := NewPluginImageResolver()
|
||||
defer resolver.Close()
|
||||
broken := &scriptedImageSource{err: status.Error(codes.Unimplemented, "method ResolveImageURLs not implemented")}
|
||||
working := &scriptedImageSource{urls: map[string]string{"poster.jpg": "working"}}
|
||||
resolver.ReplaceSources(tt.registrations(broken, working))
|
||||
|
||||
got := resolver.ResolveImageURL(context.Background(), "tmdb://poster.jpg", "card")
|
||||
if got != "working:card" {
|
||||
t.Fatalf("resolved URL = %q, want fallback working source", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginImageResolverFallsThroughForPartialBatchResults(t *testing.T) {
|
||||
resolver := NewPluginImageResolver()
|
||||
defer resolver.Close()
|
||||
|
||||
primary := &scriptedImageSource{urls: map[string]string{"a.jpg": "primary-a"}}
|
||||
secondary := &scriptedImageSource{urls: map[string]string{
|
||||
"a.jpg": "secondary-a",
|
||||
"b.jpg": "secondary-b",
|
||||
}}
|
||||
resolver.ReplaceSources([]PluginImageResolverSourceRegistration{
|
||||
{Scheme: "tmdb", Source: primary, Kind: PluginImageResolverSourceExplicit, Priority: 100, InstallationID: 1, CapabilityID: "primary"},
|
||||
{Scheme: "tmdb", Source: secondary, Kind: PluginImageResolverSourceExplicit, Priority: 10, InstallationID: 2, CapabilityID: "secondary"},
|
||||
})
|
||||
|
||||
got := resolver.ResolveImageURLs(context.Background(), []string{"tmdb://a.jpg", "tmdb://b.jpg"}, "card")
|
||||
if got["tmdb://a.jpg"] != "primary-a:card" {
|
||||
t.Fatalf("a.jpg = %q, want primary result", got["tmdb://a.jpg"])
|
||||
}
|
||||
if got["tmdb://b.jpg"] != "secondary-b:card" {
|
||||
t.Fatalf("b.jpg = %q, want secondary fallback result", got["tmdb://b.jpg"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginImageResolverDoesNotCacheEmptyFailure(t *testing.T) {
|
||||
resolver := NewPluginImageResolver()
|
||||
defer resolver.Close()
|
||||
|
||||
source := &scriptedImageSource{err: status.Error(codes.Unavailable, "temporary outage")}
|
||||
resolver.ReplaceSources([]PluginImageResolverSourceRegistration{
|
||||
{Scheme: "tmdb", Source: source, Kind: PluginImageResolverSourceExplicit, Priority: 100, InstallationID: 1, CapabilityID: "tmdb"},
|
||||
})
|
||||
|
||||
if got := resolver.ResolveImageURL(context.Background(), "tmdb://poster.jpg", "card"); got != "" {
|
||||
t.Fatalf("first resolved URL = %q, want empty during failure", got)
|
||||
}
|
||||
source.err = nil
|
||||
source.urls = map[string]string{"poster.jpg": "recovered"}
|
||||
|
||||
got := resolver.ResolveImageURL(context.Background(), "tmdb://poster.jpg", "card")
|
||||
if got != "recovered:card" {
|
||||
t.Fatalf("second resolved URL = %q, want recovered result", got)
|
||||
}
|
||||
if calls := source.calls.Load(); calls != 2 {
|
||||
t.Fatalf("source calls = %d, want 2 to prove failure was not cached", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPluginImageResolverCoalescesConcurrentBatchMisses(t *testing.T) {
|
||||
expiresAt := time.Now().Add(time.Hour)
|
||||
source := &fakeExpiringImageSource{expiresAt: &expiresAt, delay: 50 * time.Millisecond}
|
||||
|
||||
@@ -34,6 +34,11 @@ type MetadataProviderClient struct {
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
type ImageResolverClient struct {
|
||||
client pluginv1.ImageResolverClient
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
type MarkerProviderClient struct {
|
||||
client pluginv1.MarkerProviderClient
|
||||
timeout time.Duration
|
||||
@@ -113,6 +118,16 @@ func (c *Client) MetadataProvider(capabilityID string) (*MetadataProviderClient,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) ImageResolver(capabilityID string) (*ImageResolverClient, error) {
|
||||
if err := c.requireCapability("image_resolver.v1", capabilityID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ImageResolverClient{
|
||||
client: c.rpc.ImageResolver(),
|
||||
timeout: DefaultMetadataTimeout,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) MarkerProvider(capabilityID string) (*MarkerProviderClient, error) {
|
||||
if err := c.requireCapability("marker_provider.v1", capabilityID); err != nil {
|
||||
return nil, err
|
||||
@@ -262,6 +277,18 @@ func (c *MetadataProviderClient) ResolveImageURLs(ctx context.Context, req *plug
|
||||
return c.client.ResolveImageURLs(callCtx, req)
|
||||
}
|
||||
|
||||
func (c *ImageResolverClient) ResolveImageURL(ctx context.Context, req *pluginv1.ResolveImageURLRequest) (*pluginv1.ResolveImageURLResponse, error) {
|
||||
callCtx, cancel := ensureDeadline(ctx, c.timeout)
|
||||
defer cancel()
|
||||
return c.client.ResolveImageURL(callCtx, req)
|
||||
}
|
||||
|
||||
func (c *ImageResolverClient) ResolveImageURLs(ctx context.Context, req *pluginv1.ResolveImageURLsRequest) (*pluginv1.ResolveImageURLsResponse, error) {
|
||||
callCtx, cancel := ensureDeadline(ctx, c.timeout)
|
||||
defer cancel()
|
||||
return c.client.ResolveImageURLs(callCtx, req)
|
||||
}
|
||||
|
||||
func (c *MarkerProviderClient) FetchMarkers(ctx context.Context, req *pluginv1.FetchMarkersRequest) (*pluginv1.FetchMarkersResponse, error) {
|
||||
callCtx, cancel := ensureDeadline(ctx, c.timeout)
|
||||
defer cancel()
|
||||
|
||||
@@ -144,3 +144,42 @@ func TestClient_MarkerProvider_CapabilityGate(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestClient_ImageResolver_CapabilityGate(t *testing.T) {
|
||||
t.Run("absent capability returns error", func(t *testing.T) {
|
||||
c := makeTestClient(t, nil)
|
||||
_, err := c.ImageResolver("missing")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for missing image_resolver.v1 capability, got nil")
|
||||
}
|
||||
if !errors.Is(err, ErrCapabilityNotFound) {
|
||||
t.Errorf("expected ErrCapabilityNotFound, got %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("declared capability returns no error", func(t *testing.T) {
|
||||
c := makeTestClient(t, []*pluginv1.CapabilityDescriptor{
|
||||
{Type: "image_resolver.v1", Id: "tmdb"},
|
||||
})
|
||||
got, err := c.ImageResolver("tmdb")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatal("expected non-nil ImageResolverClient")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("wrong id returns error", func(t *testing.T) {
|
||||
c := makeTestClient(t, []*pluginv1.CapabilityDescriptor{
|
||||
{Type: "image_resolver.v1", Id: "tmdb"},
|
||||
})
|
||||
_, err := c.ImageResolver("imdb")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for mismatched image_resolver.v1 capability id, got nil")
|
||||
}
|
||||
if !errors.Is(err, ErrCapabilityNotFound) {
|
||||
t.Errorf("expected ErrCapabilityNotFound, got %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
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)
|
||||
@@ -450,6 +451,18 @@ func (s *Service) MetadataProviderClient(
|
||||
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,
|
||||
|
||||
@@ -201,6 +201,10 @@ func (f *fakePluginClient) MetadataProvider(string) (*pluginhost.MetadataProvide
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (f *fakePluginClient) ImageResolver(string) (*pluginhost.ImageResolverClient, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (f *fakePluginClient) MarkerProvider(string) (*pluginhost.MarkerProviderClient, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user