package pluginhost import ( "context" "errors" "fmt" "sync" "time" "google.golang.org/protobuf/proto" pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1" sdkruntime "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginsdk/runtime" ) var ( ErrClientNotFound = errors.New("plugin client not found") ErrCapabilityNotFound = errors.New("plugin capability not found") ErrPluginUnhealthy = errors.New("plugin is unhealthy") ) type Client struct { installationID int manifest *pluginv1.PluginManifest rpc *sdkruntime.Client capabilities map[string]*pluginv1.CapabilityDescriptor mu sync.RWMutex unhealthy bool } type MetadataProviderClient struct { client pluginv1.MetadataProviderClient timeout time.Duration } type ImageResolverClient struct { client pluginv1.ImageResolverClient timeout time.Duration } type MarkerProviderClient struct { client pluginv1.MarkerProviderClient timeout time.Duration } type MediaAnalyzerClient struct { client pluginv1.MediaAnalyzerClient timeout time.Duration } type ScheduledTaskClient struct { client pluginv1.ScheduledTaskClient timeout time.Duration } type ScanSourceClient struct { client pluginv1.ScanSourceClient timeout time.Duration } type RequestRouterClient struct { client pluginv1.RequestRouterClient timeout time.Duration } type EventConsumerClient struct { client pluginv1.EventConsumerClient timeout time.Duration } type AuthProviderClient struct { client pluginv1.AuthProviderClient timeout time.Duration } type HTTPRoutesClient struct { client pluginv1.HttpRoutesClient timeout time.Duration } type WatchSyncProviderClient struct { client pluginv1.WatchSyncProviderClient deviceAuthorization pluginv1.WatchSyncDeviceAuthorizationServiceClient timeout time.Duration } func newClient(installationID int, rpc *sdkruntime.Client, manifest *pluginv1.PluginManifest) *Client { capabilities := make(map[string]*pluginv1.CapabilityDescriptor, len(manifest.GetCapabilities())) for _, capability := range manifest.GetCapabilities() { capabilities[capabilityKey(capability.GetType(), capability.GetId())] = capability } return &Client{ installationID: installationID, manifest: proto.Clone(manifest).(*pluginv1.PluginManifest), rpc: rpc, capabilities: capabilities, } } func (c *Client) Manifest() *pluginv1.PluginManifest { return proto.Clone(c.manifest).(*pluginv1.PluginManifest) } func (c *Client) Capabilities() []*pluginv1.CapabilityDescriptor { c.mu.RLock() defer c.mu.RUnlock() result := make([]*pluginv1.CapabilityDescriptor, 0, len(c.capabilities)) for _, capability := range c.capabilities { result = append(result, proto.Clone(capability).(*pluginv1.CapabilityDescriptor)) } return result } func (c *Client) MetadataProvider(capabilityID string) (*MetadataProviderClient, error) { if err := c.requireCapability("metadata_provider.v1", capabilityID); err != nil { return nil, err } return &MetadataProviderClient{ client: c.rpc.MetadataProvider(), timeout: DefaultMetadataTimeout, }, 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 } return &MarkerProviderClient{ client: c.rpc.MarkerProvider(), timeout: DefaultMarkerProviderTimeout, }, nil } func (c *Client) MediaAnalyzer(capabilityID string) (*MediaAnalyzerClient, error) { if err := c.requireCapability("media_analyzer.v1", capabilityID); err != nil { return nil, err } return &MediaAnalyzerClient{ client: c.rpc.MediaAnalyzer(), timeout: DefaultAnalyzerTimeout, }, nil } func (c *Client) ScheduledTask(capabilityID string) (*ScheduledTaskClient, error) { if err := c.requireCapability("scheduled_task.v1", capabilityID); err != nil { return nil, err } return &ScheduledTaskClient{ client: c.rpc.ScheduledTask(), timeout: DefaultControlTimeout, }, nil } func (c *Client) ScanSource(capabilityID string) (*ScanSourceClient, error) { if err := c.requireCapability("scan_source.v1", capabilityID); err != nil { return nil, err } return &ScanSourceClient{ client: c.rpc.ScanSource(), timeout: DefaultScanSourceTimeout, }, nil } func (c *Client) RequestRouter(capabilityID string) (*RequestRouterClient, error) { if err := c.requireCapability("request_router.v1", capabilityID); err != nil { return nil, err } return &RequestRouterClient{ client: c.rpc.RequestRouter(), timeout: DefaultRequestRouterTimeout, }, nil } func (c *Client) EventConsumer(capabilityID string) (*EventConsumerClient, error) { if err := c.requireCapability("event_consumer.v1", capabilityID); err != nil { return nil, err } return &EventConsumerClient{ client: c.rpc.EventConsumer(), timeout: DefaultEventTimeout, }, nil } func (c *Client) AuthProvider(capabilityID string) (*AuthProviderClient, error) { if err := c.requireCapability("auth_provider.v1", capabilityID); err != nil { return nil, err } return &AuthProviderClient{ client: c.rpc.AuthProvider(), timeout: DefaultAuthTimeout, }, nil } func (c *Client) HTTPRoutes(capabilityID string) (*HTTPRoutesClient, error) { if err := c.requireCapability("http_routes.v1", capabilityID); err != nil { return nil, err } return &HTTPRoutesClient{ client: c.rpc.HttpRoutes(), timeout: DefaultRouteTimeout, }, nil } func (c *Client) WatchSyncProvider(capabilityID string) (*WatchSyncProviderClient, error) { if err := c.requireCapability("watch_sync_provider.v1", capabilityID); err != nil { return nil, err } return &WatchSyncProviderClient{ client: c.rpc.WatchSyncProvider(), deviceAuthorization: c.rpc.WatchSyncDeviceAuthorization(), timeout: DefaultWatchSyncTimeout, }, nil } func (c *Client) markUnhealthy() { c.mu.Lock() defer c.mu.Unlock() c.unhealthy = true } func (c *Client) requireCapability(capabilityType, capabilityID string) error { c.mu.RLock() defer c.mu.RUnlock() if c.unhealthy { return ErrPluginUnhealthy } descriptor, ok := c.capabilities[capabilityKey(capabilityType, capabilityID)] if !ok || descriptor == nil { return fmt.Errorf("%w: %s/%s", ErrCapabilityNotFound, capabilityType, capabilityID) } return nil } func (c *MetadataProviderClient) Search(ctx context.Context, req *pluginv1.SearchMetadataRequest) (*pluginv1.SearchMetadataResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Search(callCtx, req) } func (c *MetadataProviderClient) GetMetadata(ctx context.Context, req *pluginv1.GetMetadataRequest) (*pluginv1.GetMetadataResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetMetadata(callCtx, req) } func (c *MetadataProviderClient) GetPersonDetail(ctx context.Context, req *pluginv1.GetPersonDetailRequest) (*pluginv1.GetPersonDetailResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetPersonDetail(callCtx, req) } func (c *MetadataProviderClient) GetSeasons(ctx context.Context, req *pluginv1.GetSeasonsRequest) (*pluginv1.GetSeasonsResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetSeasons(callCtx, req) } func (c *MetadataProviderClient) GetEpisodes(ctx context.Context, req *pluginv1.GetEpisodesRequest) (*pluginv1.GetEpisodesResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetEpisodes(callCtx, req) } func (c *MetadataProviderClient) GetImages(ctx context.Context, req *pluginv1.GetImagesRequest) (*pluginv1.GetImagesResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetImages(callCtx, req) } func (c *MetadataProviderClient) 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 *MetadataProviderClient) 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 *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() return c.client.FetchMarkers(callCtx, req) } func (c *MarkerProviderClient) SubmitMarker(ctx context.Context, req *pluginv1.SubmitMarkerRequest) (*pluginv1.SubmitMarkerResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.SubmitMarker(callCtx, req) } func (c *MarkerProviderClient) GetMarkerProviderStats(ctx context.Context, req *pluginv1.GetMarkerProviderStatsRequest) (*pluginv1.MarkerProviderStatsResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetMarkerProviderStats(callCtx, req) } func (c *MediaAnalyzerClient) Analyze(ctx context.Context, req *pluginv1.AnalyzeMediaRequest) (*pluginv1.AnalyzeMediaResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Analyze(callCtx, req) } func (c *ScheduledTaskClient) Run(ctx context.Context, req *pluginv1.RunScheduledTaskRequest) (*pluginv1.RunScheduledTaskResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Run(callCtx, req) } func (c *ScanSourceClient) PollChanges(ctx context.Context, req *pluginv1.PollChangesRequest) (*pluginv1.PollChangesResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.PollChanges(callCtx, req) } func (c *RequestRouterClient) Fulfill(ctx context.Context, req *pluginv1.FulfillRequest) (*pluginv1.FulfillResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Fulfill(callCtx, req) } func (c *RequestRouterClient) CheckStatus(ctx context.Context, req *pluginv1.CheckStatusRequest) (*pluginv1.CheckStatusResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.CheckStatus(callCtx, req) } func (c *RequestRouterClient) ListConfigOptions(ctx context.Context, req *pluginv1.ListConfigOptionsRequest) (*pluginv1.ListConfigOptionsResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.ListConfigOptions(callCtx, req) } func (c *RequestRouterClient) TestConnection(ctx context.Context, req *pluginv1.TestConnectionRequest) (*pluginv1.TestConnectionResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.TestConnection(callCtx, req) } func (c *RequestRouterClient) Validate(ctx context.Context, req *pluginv1.ValidateRequest) (*pluginv1.ValidateResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Validate(callCtx, req) } func (c *EventConsumerClient) HandleEvent(ctx context.Context, req *pluginv1.HandleEventRequest) (*pluginv1.HandleEventResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.HandleEvent(callCtx, req) } func (c *AuthProviderClient) Authenticate(ctx context.Context, req *pluginv1.AuthenticateRequest) (*pluginv1.AuthenticateResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Authenticate(callCtx, req) } func (c *AuthProviderClient) InitAuthorize(ctx context.Context, req *pluginv1.InitAuthorizeRequest) (*pluginv1.InitAuthorizeResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.InitAuthorize(callCtx, req) } func (c *AuthProviderClient) ExchangeCode(ctx context.Context, req *pluginv1.ExchangeCodeRequest) (*pluginv1.AuthenticateResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.ExchangeCode(callCtx, req) } func (c *HTTPRoutesClient) Handle(ctx context.Context, req *pluginv1.HandleHTTPRequest) (*pluginv1.HandleHTTPResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.Handle(callCtx, req) } func (c *WatchSyncProviderClient) ExchangeAPIKey(ctx context.Context, req *pluginv1.WatchSyncExchangeAPIKeyRequest) (*pluginv1.WatchSyncCredentialResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.ExchangeAPIKey(callCtx, req) } func (c *WatchSyncProviderClient) StartDeviceAuthorization(ctx context.Context, req *pluginv1.WatchSyncDeviceAuthorizationServiceStartRequest) (*pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.deviceAuthorization.Start(callCtx, req) } func (c *WatchSyncProviderClient) PollDeviceAuthorization(ctx context.Context, req *pluginv1.WatchSyncDeviceAuthorizationServicePollRequest) (*pluginv1.WatchSyncDeviceAuthorizationServicePollResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.deviceAuthorization.Poll(callCtx, req) } func (c *WatchSyncProviderClient) RefreshCredentials(ctx context.Context, req *pluginv1.WatchSyncRefreshCredentialsRequest) (*pluginv1.WatchSyncCredentialResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.RefreshCredentials(callCtx, req) } func (c *WatchSyncProviderClient) GetAccount(ctx context.Context, req *pluginv1.WatchSyncGetAccountRequest) (*pluginv1.WatchSyncGetAccountResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.GetAccount(callCtx, req) } func (c *WatchSyncProviderClient) ApplyEvents(ctx context.Context, req *pluginv1.WatchSyncApplyEventsRequest) (*pluginv1.WatchSyncApplyEventsResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.ApplyEvents(callCtx, req) } func (c *WatchSyncProviderClient) ListRemoteState(ctx context.Context, req *pluginv1.WatchSyncListRemoteStateRequest) (*pluginv1.WatchSyncListRemoteStateResponse, error) { callCtx, cancel := ensureDeadline(ctx, c.timeout) defer cancel() return c.client.ListRemoteState(callCtx, req) } func ensureDeadline(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { if _, ok := ctx.Deadline(); ok { return ctx, func() {} } return context.WithTimeout(ctx, timeout) } func capabilityKey(capabilityType, capabilityID string) string { return capabilityType + "\x00" + capabilityID }