silo-server/internal/watchsync/plugin_provider.go

package watchsync

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode"
	"unicode/utf8"

	pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
	publicconfig "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginsdk/config"
	"github.com/Silo-Server/silo-server/internal/historyimport"
	hostplugins "github.com/Silo-Server/silo-server/internal/plugins"
	"github.com/Silo-Server/silo-server/internal/userstore"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type WatchSyncPluginClient interface {
	ExchangeAPIKey(context.Context, *pluginv1.WatchSyncExchangeAPIKeyRequest) (*pluginv1.WatchSyncCredentialResponse, error)
	StartDeviceAuthorization(context.Context, *pluginv1.WatchSyncDeviceAuthorizationServiceStartRequest) (*pluginv1.WatchSyncDeviceAuthorizationServiceStartResponse, error)
	PollDeviceAuthorization(context.Context, *pluginv1.WatchSyncDeviceAuthorizationServicePollRequest) (*pluginv1.WatchSyncDeviceAuthorizationServicePollResponse, error)
	RefreshCredentials(context.Context, *pluginv1.WatchSyncRefreshCredentialsRequest) (*pluginv1.WatchSyncCredentialResponse, error)
	GetAccount(context.Context, *pluginv1.WatchSyncGetAccountRequest) (*pluginv1.WatchSyncGetAccountResponse, error)
	ApplyEvents(context.Context, *pluginv1.WatchSyncApplyEventsRequest) (*pluginv1.WatchSyncApplyEventsResponse, error)
	ListRemoteState(context.Context, *pluginv1.WatchSyncListRemoteStateRequest) (*pluginv1.WatchSyncListRemoteStateResponse, error)
}

type WatchSyncPluginClientResolver func(context.Context, int, string) (WatchSyncPluginClient, error)
type WatchSyncPluginConfigResolver func(context.Context, int) (*pluginv1.WatchSyncProviderConfig, error)

type PluginCredentialRepository interface {
	UpsertConnection(context.Context, Connection) (Connection, error)
}

type PluginProviderOptions struct {
	InstallationID         int
	ProviderKey            string
	CapabilityID           string
	DisplayName            string
	Descriptor             *pluginv1.WatchSyncProviderDescriptor
	ConnectionConfigSchema []*pluginv1.ConfigSchema
	ResolveClient          WatchSyncPluginClientResolver
	ResolveConfig          WatchSyncPluginConfigResolver
	Repository             PluginCredentialRepository
}

type PluginProvider struct {
	installationID         int
	providerKey            string
	capabilityID           string
	displayName            string
	descriptor             *pluginv1.WatchSyncProviderDescriptor
	connectionConfigSchema []*pluginv1.ConfigSchema
	authMethod             string
	supportedMedia         map[pluginv1.WatchSyncMediaType]struct{}
	resolveClient          WatchSyncPluginClientResolver
	resolveConfig          WatchSyncPluginConfigResolver
	repository             PluginCredentialRepository
}

const (
	watchSyncUnsupportedMovieMediaMessage   = "watch sync plugin does not support movie media"
	watchSyncUnsupportedEpisodeMediaMessage = "watch sync plugin does not support episode media"
	watchSyncUnsupportedMediaMessage        = "watch sync plugin does not support this media type"
	watchSyncJSONSchemaNumberType           = "number"
	watchSyncJSONSchemaBooleanType          = "boolean"
)

func NewPluginProvider(options PluginProviderOptions) (*PluginProvider, error) {
	if options.InstallationID <= 0 || strings.TrimSpace(options.CapabilityID) == "" {
		return nil, fmt.Errorf("watch sync plugin installation and capability are required")
	}
	if strings.TrimSpace(options.ProviderKey) == "" {
		return nil, fmt.Errorf("watch sync plugin provider key is required")
	}
	if options.Descriptor == nil {
		return nil, fmt.Errorf("watch sync plugin descriptor is required")
	}
	authMethod, err := supportedWatchSyncAuthMethod(options.Descriptor)
	if err != nil {
		return nil, fmt.Errorf("watch sync plugin %q %w", options.ProviderKey, err)
	}
	supportedMedia, err := supportedWatchSyncMediaTypes(options.Descriptor)
	if err != nil {
		return nil, fmt.Errorf("watch sync plugin %q %w", options.ProviderKey, err)
	}
	if err := validateWatchSyncConnectionConfigSchemas(options.ConnectionConfigSchema); err != nil {
		return nil, fmt.Errorf("watch sync plugin %q %w", options.ProviderKey, err)
	}
	if authMethod != AuthMethodAPIKey && hasWatchSyncConnectionConfigSchema(options.ConnectionConfigSchema) {
		return nil, fmt.Errorf("watch sync plugin %q connection config requires API-key authentication", options.ProviderKey)
	}
	if options.ResolveClient == nil {
		return nil, fmt.Errorf("watch sync plugin client resolver is required")
	}
	return &PluginProvider{
		installationID:         options.InstallationID,
		providerKey:            options.ProviderKey,
		capabilityID:           options.CapabilityID,
		displayName:            options.DisplayName,
		descriptor:             options.Descriptor,
		connectionConfigSchema: append([]*pluginv1.ConfigSchema(nil), options.ConnectionConfigSchema...),
		authMethod:             authMethod,
		supportedMedia:         supportedMedia,
		resolveClient:          options.ResolveClient,
		resolveConfig:          options.ResolveConfig,
		repository:             options.Repository,
	}, nil
}

func hasWatchSyncConnectionConfigSchema(schemas []*pluginv1.ConfigSchema) bool {
	for _, schema := range schemas {
		if schema != nil {
			return true
		}
	}
	return false
}

func (p *PluginProvider) Key() string { return p.providerKey }

func (p *PluginProvider) DisplayName() string {
	if strings.TrimSpace(p.displayName) != "" {
		return p.displayName
	}
	return p.capabilityID
}

func (p *PluginProvider) ProviderSource() string { return providerSourcePlugin }

// HistorySource is provider-specific so importing from one plugin suppresses
// only the echo back to that same connection. A shared generic source would
// also suppress legitimate Floppy-to-Trakt (or other cross-provider) sync.
func (p *PluginProvider) HistorySource() userstore.WatchHistorySource {
	return userstore.WatchHistorySource(p.providerKey)
}

func (p *PluginProvider) AuthMethod() string { return p.authMethod }

func (p *PluginProvider) ConnectionConfigSchema() []hostplugins.ConfigSchemaView {
	return hostplugins.ConfigSchemaViews(p.connectionConfigSchema)
}

func (p *PluginProvider) usesHostPluginConfig() {}

func (p *PluginProvider) authoritativeRefreshProvider() {}

func (p *PluginProvider) ExportBatchSize() int {
	if size := int(p.descriptor.GetMaxBatchSize()); size > 0 {
		return size
	}
	return 1
}

func (p *PluginProvider) Capabilities() Capabilities {
	return Capabilities{
		ImportWatched:          p.descriptor.GetImportWatched(),
		ImportProgress:         p.descriptor.GetImportProgress(),
		ExportWatched:          p.descriptor.GetExportWatched(),
		ExportUnwatched:        p.descriptor.GetExportUnwatched(),
		ImportFavorites:        p.descriptor.GetImportFavorites(),
		ExportFavorites:        p.descriptor.GetExportFavorites(),
		RemoveFavorites:        p.descriptor.GetRemoveFavorites(),
		ImportWatchlist:        p.descriptor.GetImportWatchlist(),
		ExportWatchlist:        p.descriptor.GetExportWatchlist(),
		RemoveWatchlist:        p.descriptor.GetRemoveWatchlist(),
		ProvidesWatchlistOrder: p.descriptor.GetProvidesWatchlistOrder(),
		ScrobblePlayback:       p.descriptor.GetScrobblePlayback(),
	}
}

func (p *PluginProvider) ConnectWithAPIKey(ctx context.Context, apiKey string) (TokenSet, ProviderAccount, error) {
	return p.ConnectWithAPIKeyConfig(ctx, apiKey, nil)
}

func (p *PluginProvider) ConnectWithAPIKeyConfig(
	ctx context.Context,
	apiKey string,
	connectionConfig ConnectionConfigValues,
) (TokenSet, ProviderAccount, error) {
	if p.authMethod != AuthMethodAPIKey {
		return TokenSet{}, ProviderAccount{}, errors.New("watch sync plugin does not support API-key authentication")
	}
	config, err := p.providerConfig(ctx)
	if err != nil {
		return TokenSet{}, ProviderAccount{}, err
	}
	connectionValues, connectionSecrets, err := p.connectionConfig(connectionConfig)
	if err != nil {
		return TokenSet{}, ProviderAccount{}, err
	}
	config = mergeWatchSyncProviderConfig(config, connectionValues)
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return TokenSet{}, ProviderAccount{}, watchSyncUnavailableError()
	}
	response, err := client.ExchangeAPIKey(ctx, &pluginv1.WatchSyncExchangeAPIKeyRequest{
		CapabilityId:   p.capabilityID,
		ProviderConfig: config,
		ApiKey:         apiKey,
	})
	if err != nil {
		return TokenSet{}, ProviderAccount{}, watchSyncRPCError()
	}
	faultSecrets := append([]string{apiKey}, connectionSecrets...)
	if err := watchSyncFaultError(p.Key(), response.GetFault(), faultSecrets...); err != nil {
		return TokenSet{}, ProviderAccount{}, err
	}
	tokens, err := tokenSetFromProto(response.GetCredentials())
	if err != nil {
		return TokenSet{}, ProviderAccount{}, err
	}
	account, err := accountFromProto(response.GetAccount())
	if err != nil {
		return TokenSet{}, ProviderAccount{}, err
	}
	return tokens, account, nil
}

func (p *PluginProvider) StartDeviceAuth(ctx context.Context, _ ServerConfig) (DeviceAuthSession, error) {
	if p.authMethod != AuthMethodDeviceCode {
		return DeviceAuthSession{}, errors.New("watch sync plugin does not support device authorization")
	}
	config, err := p.providerConfig(ctx)
	if err != nil {
		return DeviceAuthSession{}, err
	}
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return DeviceAuthSession{}, watchSyncUnavailableError()
	}
	response, err := client.StartDeviceAuthorization(ctx, &pluginv1.WatchSyncDeviceAuthorizationServiceStartRequest{
		CapabilityId:   p.capabilityID,
		ProviderConfig: config,
	})
	if err != nil {
		return DeviceAuthSession{}, watchSyncRPCError()
	}
	if err := watchSyncFaultError(p.Key(), response.GetFault()); err != nil {
		return DeviceAuthSession{}, err
	}
	if strings.TrimSpace(response.GetUserCode()) == "" || len(response.GetProviderState()) == 0 || response.GetExpiresAt() == nil {
		return DeviceAuthSession{}, errors.New("watch sync plugin returned an incomplete device authorization")
	}
	if err := response.GetExpiresAt().CheckValid(); err != nil || !response.GetExpiresAt().AsTime().After(time.Now()) {
		return DeviceAuthSession{}, errors.New("watch sync plugin returned an invalid device authorization expiry")
	}
	interval := 5
	if pollingInterval := response.GetPollingInterval(); pollingInterval != nil {
		if err := pollingInterval.CheckValid(); err != nil || pollingInterval.AsDuration() <= 0 {
			return DeviceAuthSession{}, errors.New("watch sync plugin returned an invalid device authorization polling interval")
		}
		interval = max(1, int(pollingInterval.AsDuration().Seconds()))
	}
	verificationURL, err := validDeviceVerificationURL(response.GetVerificationUrl())
	if err != nil {
		return DeviceAuthSession{}, err
	}
	if complete := strings.TrimSpace(response.GetVerificationUrlComplete()); complete != "" {
		verificationURL, err = validDeviceVerificationURL(complete)
		if err != nil {
			return DeviceAuthSession{}, err
		}
	}
	return DeviceAuthSession{
		DeviceCode:      base64.RawURLEncoding.EncodeToString(response.GetProviderState()),
		UserCode:        strings.TrimSpace(response.GetUserCode()),
		VerificationURL: verificationURL,
		IntervalSeconds: interval,
		ExpiresAt:       response.GetExpiresAt().AsTime(),
	}, nil
}

func (p *PluginProvider) PollDeviceAuth(ctx context.Context, _ ServerConfig, session DeviceAuthSession) (TokenSet, error) {
	state, err := base64.RawURLEncoding.DecodeString(session.DeviceCode)
	if err != nil {
		return TokenSet{}, errors.New("watch sync plugin device authorization state is invalid")
	}
	config, err := p.providerConfig(ctx)
	if err != nil {
		return TokenSet{}, err
	}
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return TokenSet{}, watchSyncUnavailableError()
	}
	response, err := client.PollDeviceAuthorization(ctx, &pluginv1.WatchSyncDeviceAuthorizationServicePollRequest{
		CapabilityId:   p.capabilityID,
		ProviderConfig: config,
		ProviderState:  state,
	})
	if err != nil {
		return TokenSet{}, watchSyncRPCError()
	}
	if err := watchSyncFaultError(p.Key(), response.GetFault()); err != nil {
		return TokenSet{}, err
	}
	switch response.GetStatus() {
	case pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_PENDING:
		updated, err := updatedPendingDeviceAuthSession(session, response)
		if err != nil {
			return TokenSet{}, err
		}
		return TokenSet{}, deviceAuthorizationPendingError{session: updated}
	case pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_DENIED:
		return TokenSet{}, errors.New("watch sync plugin device authorization was denied")
	case pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_EXPIRED:
		return TokenSet{}, errors.New("watch sync plugin device authorization expired")
	case pluginv1.WatchSyncDeviceAuthorizationStatus_WATCH_SYNC_DEVICE_AUTHORIZATION_STATUS_AUTHORIZED:
		return tokenSetFromProto(response.GetCredentials())
	default:
		return TokenSet{}, errors.New("watch sync plugin returned an invalid device authorization status")
	}
}

func validDeviceVerificationURL(value string) (string, error) {
	value = strings.TrimSpace(value)
	parsed, err := url.Parse(value)
	if err != nil || parsed == nil || !parsed.IsAbs() || parsed.Host == "" || parsed.User != nil ||
		(parsed.Scheme != "http" && parsed.Scheme != "https") {
		return "", errors.New("watch sync plugin returned an invalid device authorization verification URL")
	}
	return parsed.String(), nil
}

func updatedPendingDeviceAuthSession(
	session DeviceAuthSession,
	response *pluginv1.WatchSyncDeviceAuthorizationServicePollResponse,
) (DeviceAuthSession, error) {
	if response.ProviderState != nil {
		session.DeviceCode = base64.RawURLEncoding.EncodeToString(response.GetProviderState())
	}
	if pollingInterval := response.GetPollingInterval(); pollingInterval != nil {
		if err := pollingInterval.CheckValid(); err != nil || pollingInterval.AsDuration() <= 0 {
			return DeviceAuthSession{}, errors.New("watch sync plugin returned an invalid device authorization polling interval")
		}
		session.IntervalSeconds = max(1, int(pollingInterval.AsDuration().Seconds()))
	}
	if expiresAt := response.GetExpiresAt(); expiresAt != nil {
		if err := expiresAt.CheckValid(); err != nil || !expiresAt.AsTime().After(time.Now()) {
			return DeviceAuthSession{}, errors.New("watch sync plugin returned an invalid device authorization expiry")
		}
		session.ExpiresAt = expiresAt.AsTime()
	}
	return session, nil
}

func (p *PluginProvider) RefreshToken(ctx context.Context, _ ServerConfig, conn Connection) (TokenSet, error) {
	authContext, err := p.authenticatedContext(ctx, conn)
	if err != nil {
		return TokenSet{}, err
	}
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return TokenSet{}, watchSyncUnavailableError()
	}
	response, err := client.RefreshCredentials(ctx, &pluginv1.WatchSyncRefreshCredentialsRequest{
		Context: authContext,
	})
	if err != nil {
		return TokenSet{}, watchSyncRPCError()
	}
	var tokens TokenSet
	if response.GetCredentials() != nil {
		tokens, err = tokenSetFromProto(response.GetCredentials())
		if err != nil {
			return TokenSet{}, err
		}
	}
	if err := watchSyncFaultError(p.Key(), response.GetFault(), conn.AccessToken, conn.RefreshToken, tokens.AccessToken, tokens.RefreshToken); err != nil {
		return tokens, err
	}
	if response.GetCredentials() == nil {
		return TokenSet{}, errors.New("watch sync plugin returned no access token")
	}
	return tokens, nil
}

func (p *PluginProvider) LookupAccount(ctx context.Context, _ ServerConfig, conn Connection) (ProviderAccount, error) {
	authContext, err := p.authenticatedContext(ctx, conn)
	if err != nil {
		return ProviderAccount{}, err
	}
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return ProviderAccount{}, watchSyncUnavailableError()
	}
	response, err := client.GetAccount(ctx, &pluginv1.WatchSyncGetAccountRequest{
		Context: authContext,
	})
	if err != nil {
		return ProviderAccount{}, watchSyncRPCError()
	}
	if err := watchSyncFaultError(p.Key(), response.GetFault(), conn.AccessToken, conn.RefreshToken); err != nil {
		return ProviderAccount{}, err
	}
	return accountFromProto(response.GetAccount())
}

func (p *PluginProvider) FetchHistory(context.Context, ServerConfig, Connection) ([]RemotePlay, error) {
	// A desired-state tracker is not a timestamped play-history source. Silo's
	// durable local export rows provide reconciliation for this provider.
	return nil, nil
}

func (p *PluginProvider) ExportHistory(ctx context.Context, _ ServerConfig, conn Connection, plays []LocalPlay) (ExportResult, error) {
	result := ExportResult{Failed: map[string]string{}}
	if len(plays) == 0 {
		return result, nil
	}
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return result, watchSyncUnavailableError()
	}

	// Perform one bounded RPC per service iteration. This lets the existing
	// exporter commit every per-event result before requesting the next batch,
	// and avoids holding one sync run across many sequential plugin deadlines.
	batchSize := len(plays)
	if maximum := p.ExportBatchSize(); batchSize > maximum {
		batchSize = maximum
	}
	events := make([]*pluginv1.WatchSyncEvent, 0, batchSize)
	selectedPlays := make([]LocalPlay, 0, batchSize)
	for _, play := range plays[:batchSize] {
		event := watchEventFromLocalPlay(play, pluginv1.WatchSyncOrigin_WATCH_SYNC_ORIGIN_RECONCILIATION)
		if !p.supportsMedia(event.GetMedia().GetMediaType()) {
			result.Failed[play.HistoryID] = unsupportedWatchSyncMediaMessage(event.GetMedia().GetMediaType())
			continue
		}
		events = append(events, event)
		selectedPlays = append(selectedPlays, play)
	}
	if len(events) == 0 {
		return result, nil
	}
	authContext, err := p.authenticatedContext(ctx, conn)
	if err != nil {
		return result, err
	}
	response, err := client.ApplyEvents(ctx, &pluginv1.WatchSyncApplyEventsRequest{
		Context: authContext,
		Events:  events,
	})
	if err != nil {
		return result, watchSyncRPCError()
	}
	if response.GetUpdatedCredentials() != nil {
		conn, err = p.persistUpdatedCredentials(ctx, conn, response.GetUpdatedCredentials())
		if err != nil {
			return result, err
		}
	}
	if err := watchSyncFaultError(p.Key(), response.GetFault(), conn.AccessToken, conn.RefreshToken); err != nil {
		// Batch-level faults apply to the whole request; no per-event results
		// are committed when the host is told to ignore them.
		return result, err
	}

	var rateLimited error
	for _, event := range events {
		apply := resultForEvent(response.GetResults(), event.GetEventId())
		if fault := apply.GetFault(); apply.GetStatus() == pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY &&
			fault != nil && fault.GetCode() == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_RATE_LIMITED {
			retry := time.Duration(0)
			if fault.GetRetryAfter() != nil {
				retry = fault.GetRetryAfter().AsDuration()
			}
			rateLimited = RateLimitedError{Provider: p.Key(), RetryAfter: retry}
			break
		}
	}
	for index, event := range events {
		historyID := selectedPlays[index].HistoryID
		apply := resultForEvent(response.GetResults(), event.GetEventId())
		switch apply.GetStatus() {
		case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED,
			pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_NO_CHANGE:
			result.Sent = append(result.Sent, historyID)
		case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_REJECTED:
			result.NotFound = append(result.NotFound, historyID)
		case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY:
			if fault := apply.GetFault(); fault != nil &&
				fault.GetCode() == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_RATE_LIMITED {
				continue
			}
			result.Failed[historyID] = safeApplyMessage(apply, conn.AccessToken, conn.RefreshToken)
		default:
			if rateLimited == nil {
				result.Failed[historyID] = "watch sync plugin omitted a valid event result"
			}
		}
	}
	return result, rateLimited
}

func (p *PluginProvider) Start(ctx context.Context, _ ServerConfig, conn Connection, event ScrobbleEvent) error {
	return p.applyScrobble(ctx, conn, event, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_START)
}

func (p *PluginProvider) Pause(ctx context.Context, _ ServerConfig, conn Connection, event ScrobbleEvent) error {
	return p.applyScrobble(ctx, conn, event, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_PAUSE)
}

func (p *PluginProvider) Stop(ctx context.Context, _ ServerConfig, conn Connection, event ScrobbleEvent) error {
	return p.applyScrobble(ctx, conn, event, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_STOP)
}

func (p *PluginProvider) applyScrobble(
	ctx context.Context,
	conn Connection,
	event ScrobbleEvent,
	operation pluginv1.WatchSyncOperation,
) error {
	watchEvent := watchEventFromScrobble(event, operation)
	if !p.supportsMedia(watchEvent.GetMedia().GetMediaType()) {
		return watchSyncProviderFaultError{message: unsupportedWatchSyncMediaMessage(watchEvent.GetMedia().GetMediaType())}
	}
	result, statuses, err := p.applyPluginEventsDetailed(ctx, conn, []*pluginv1.WatchSyncEvent{watchEvent}, []string{watchEvent.GetEventId()})
	if err != nil {
		return err
	}
	switch statuses[watchEvent.GetEventId()] {
	case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED,
		pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_NO_CHANGE:
		return nil
	case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY:
		return retryableProviderError{message: result.Failed[watchEvent.GetEventId()]}
	case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_REJECTED:
		return watchSyncProviderFaultError{
			code:    pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_PERMANENT,
			message: "watch sync plugin rejected the playback event",
		}
	default:
		return watchSyncProviderFaultError{
			code:    pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_PERMANENT,
			message: "watch sync plugin did not confirm the playback event",
		}
	}
}

func (p *PluginProvider) ScrobbleOrderingKey(conn Connection, event ScrobbleEvent) string {
	seriesID := firstNonEmptyWatchID(event.SeriesTVDBID, event.SeriesTMDBID, event.SeriesIMDbID, event.MediaItemID)
	return "plugin-watch-sync:" + conn.ID + ":" + seriesID
}

func (p *PluginProvider) authenticatedContext(ctx context.Context, conn Connection) (*pluginv1.WatchSyncAuthenticatedContext, error) {
	config, err := p.providerConfig(ctx)
	if err != nil {
		return nil, err
	}
	return &pluginv1.WatchSyncAuthenticatedContext{
		CapabilityId:   p.capabilityID,
		ProviderConfig: config,
		Credentials:    credentialsFromConnection(conn),
	}, nil
}

func (p *PluginProvider) providerConfig(ctx context.Context) (*pluginv1.WatchSyncProviderConfig, error) {
	if p.resolveConfig == nil {
		return &pluginv1.WatchSyncProviderConfig{}, nil
	}
	config, err := p.resolveConfig(ctx, p.installationID)
	if err != nil {
		return nil, retryableProviderError{message: "watch sync plugin configuration is unavailable"}
	}
	if config == nil {
		config = &pluginv1.WatchSyncProviderConfig{}
	}
	return config, nil
}

func (p *PluginProvider) connectionConfig(values ConnectionConfigValues) (*pluginv1.WatchSyncProviderConfig, []string, error) {
	declared := make(map[string]*pluginv1.ConfigSchema, len(p.connectionConfigSchema))
	for _, schema := range p.connectionConfigSchema {
		if schema != nil && strings.TrimSpace(schema.GetKey()) != "" {
			declared[schema.GetKey()] = schema
		}
	}
	secrets := connectionConfigSecrets(p.connectionConfigSchema, values)
	for key := range values {
		if _, ok := declared[key]; !ok {
			return nil, nil, sanitizedConnectionConfigError(
				fmt.Errorf("watch sync connection config key %q is not declared", key),
				secrets,
			)
		}
	}

	result := &pluginv1.WatchSyncProviderConfig{
		Values:       make(map[string]string),
		SecretValues: make(map[string]string),
	}
	flattenedFields := make(map[string]string)
	for _, schema := range p.connectionConfigSchema {
		if schema == nil || strings.TrimSpace(schema.GetKey()) == "" {
			continue
		}
		value, exists := values[schema.GetKey()]
		if !exists {
			if schema.GetRequired() {
				return nil, nil, fmt.Errorf("watch sync connection config %q is required", schema.GetKey())
			}
			continue
		}
		if err := publicconfig.ValidateValue(schema, "watch sync connection config", schema.GetKey(), value); err != nil {
			return nil, nil, sanitizedConnectionConfigError(err, secrets)
		}
		if err := validateConnectionAdminFormValue(schema, value); err != nil {
			return nil, nil, sanitizedConnectionConfigError(err, secrets)
		}
		publicFieldNames, _ := hostplugins.ConfigSchemaFieldSets(schema)
		publicFields := make(map[string]struct{}, len(publicFieldNames))
		for _, field := range publicFieldNames {
			publicFields[field] = struct{}{}
		}
		for field, raw := range value {
			rawField := field
			field = strings.TrimSpace(field)
			if field == "" {
				continue
			}
			encoded, err := connectionConfigString(raw)
			if err != nil {
				return nil, nil, sanitizedConnectionConfigError(
					fmt.Errorf("encode watch sync connection config %q.%s: %w", schema.GetKey(), field, err),
					secrets,
				)
			}
			key := schema.GetKey() + "." + field
			source := fmt.Sprintf("config %q field %q", schema.GetKey(), rawField)
			if previous, exists := flattenedFields[key]; exists {
				return nil, nil, sanitizedConnectionConfigError(
					fmt.Errorf("watch sync connection %s conflicts with %s after flattening to %q", source, previous, key),
					secrets,
				)
			}
			flattenedFields[key] = source
			if _, public := publicFields[field]; public {
				result.Values[key] = encoded
			} else {
				result.SecretValues[key] = encoded
			}
		}
	}
	return result, secrets, nil
}

func connectionConfigSecrets(schemas []*pluginv1.ConfigSchema, values ConnectionConfigValues) []string {
	var secrets []string
	for _, schema := range schemas {
		if schema == nil {
			continue
		}
		value, exists := values[schema.GetKey()]
		if !exists {
			continue
		}
		_, secretFields := hostplugins.ConfigSchemaFieldSets(schema)
		for _, field := range secretFields {
			raw, exists := value[field]
			if !exists {
				continue
			}
			if encoded, err := connectionConfigString(raw); err == nil && strings.TrimSpace(encoded) != "" {
				secrets = append(secrets, encoded)
			}
			secrets = append(secrets, connectionConfigSecretStrings(raw)...)
		}
	}
	return secrets
}

func sanitizedConnectionConfigError(err error, secrets []string) error {
	return errors.New(sanitizeWatchSyncMessage(
		err.Error(),
		"watch sync connection config is invalid",
		secrets...,
	))
}

func connectionConfigSecretStrings(value any) []string {
	switch typed := value.(type) {
	case map[string]any:
		var values []string
		for _, child := range typed {
			values = append(values, connectionConfigSecretStrings(child)...)
		}
		return values
	case []any:
		var values []string
		for _, child := range typed {
			values = append(values, connectionConfigSecretStrings(child)...)
		}
		return values
	default:
		encoded, err := connectionConfigString(typed)
		if err != nil || strings.TrimSpace(encoded) == "" {
			return nil
		}
		return []string{encoded}
	}
}

func validateWatchSyncConnectionConfigSchemas(schemas []*pluginv1.ConfigSchema) error {
	seen := make(map[string]struct{}, len(schemas))
	for _, schema := range schemas {
		if schema == nil {
			continue
		}
		key := strings.TrimSpace(schema.GetKey())
		if key == "" {
			return fmt.Errorf("connection config key is required")
		}
		if _, exists := seen[key]; exists {
			return fmt.Errorf("connection config key %q is duplicated", key)
		}
		seen[key] = struct{}{}
		for _, field := range schema.GetAdminForm().GetFields() {
			if field == nil {
				continue
			}
			if strings.TrimSpace(field.GetExclusiveGroupField()) != "" {
				return fmt.Errorf(
					"connection config %q field %q uses exclusive_group_field, which watch provider setup does not support",
					schema.GetKey(),
					field.GetKey(),
				)
			}
			if field.GetDynamicOptions() && len(field.GetOptions()) == 0 {
				return fmt.Errorf(
					"connection config %q field %q requires dynamic options, which watch provider setup does not support",
					schema.GetKey(),
					field.GetKey(),
				)
			}
			if field.GetControl() == pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT {
				for _, option := range field.GetOptions() {
					if option != nil && strings.TrimSpace(option.GetValue()) == "" {
						return fmt.Errorf(
							"connection config %q field %q has a blank select option value",
							schema.GetKey(),
							field.GetKey(),
						)
					}
				}
			}
			if pattern := field.GetValidation().GetPattern(); pattern != "" {
				if _, err := regexp.Compile(pattern); err != nil {
					return fmt.Errorf(
						"connection config %q field %q has invalid validation pattern: %w",
						schema.GetKey(),
						field.GetKey(),
						err,
					)
				}
			}
		}
		if err := validateConnectionSchemaIsRenderable(schema); err != nil {
			return err
		}
	}
	return nil
}

func validateConnectionSchemaIsRenderable(schema *pluginv1.ConfigSchema) error {
	if strings.TrimSpace(schema.GetJsonSchema()) == "" && !schema.GetRequired() {
		return nil
	}
	var document struct {
		Type       string `json:"type"`
		Properties map[string]struct {
			Type  string `json:"type"`
			Items *struct {
				Type string `json:"type"`
			} `json:"items"`
		} `json:"properties"`
	}
	if err := json.Unmarshal([]byte(schema.GetJsonSchema()), &document); err != nil {
		return fmt.Errorf("connection config %q has invalid json_schema: %w", schema.GetKey(), err)
	}
	if document.Type != "object" || document.Properties == nil {
		return fmt.Errorf("connection config %q must have a renderable object json_schema", schema.GetKey())
	}
	explicit := make(map[string]*pluginv1.AdminFormField)
	if form := schema.GetAdminForm(); form != nil {
		for _, field := range form.GetFields() {
			if field != nil {
				explicit[field.GetKey()] = field
			}
		}
	}
	for key, property := range document.Properties {
		field := explicit[key]
		switch property.Type {
		case "string", watchSyncJSONSchemaNumberType, "integer", watchSyncJSONSchemaBooleanType:
			if field != nil && !connectionAdminFieldSupportsScalarType(field, property.Type) {
				return fmt.Errorf(
					"connection config %q field %q control %q cannot emit json_schema type %q",
					schema.GetKey(),
					key,
					field.GetControl().String(),
					property.Type,
				)
			}
			continue
		case "array":
			if field != nil && field.GetControl() == pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT &&
				(property.Items == nil || property.Items.Type == "string" || property.Items.Type == watchSyncJSONSchemaNumberType ||
					property.Items.Type == "integer" || property.Items.Type == watchSyncJSONSchemaBooleanType) {
				continue
			}
		default:
			// A property whose shape comes from enum/const/$ref cannot be
			// inferred from type alone, but an explicit scalar form control is
			// still a complete input mechanism. Direct object properties remain
			// unsupported because none of these controls produces an object.
			if property.Type != "object" && connectionAdminFieldRendersValue(field) {
				continue
			}
		}
		return fmt.Errorf(
			"connection config %q property %q needs a renderable admin_form field because type %q cannot be inferred",
			schema.GetKey(),
			key,
			property.Type,
		)
	}
	return nil
}

func connectionAdminFieldSupportsScalarType(field *pluginv1.AdminFormField, propertyType string) bool {
	switch field.GetControl() {
	case pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH:
		return propertyType == watchSyncJSONSchemaBooleanType
	case pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT:
		return false
	default:
		return true
	}
}

func connectionAdminFieldRendersValue(field *pluginv1.AdminFormField) bool {
	if field == nil {
		return false
	}
	switch field.GetControl() {
	case pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT,
		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXTAREA,
		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_PASSWORD,
		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_NUMBER,
		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH,
		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT,
		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT:
		return true
	default:
		return false
	}
}

func validateConnectionAdminFormValue(schema *pluginv1.ConfigSchema, value map[string]any) error {
	if schema == nil || schema.GetAdminForm() == nil {
		return nil
	}
	for _, field := range schema.GetAdminForm().GetFields() {
		if field == nil || !connectionAdminFieldIsVisible(schema.GetAdminForm(), field, value) {
			continue
		}
		raw, exists := value[field.GetKey()]
		if !exists || connectionConfigValueIsEmpty(raw) {
			if field.GetRequired() {
				return fmt.Errorf("connection config %q field %q is required", schema.GetKey(), field.GetKey())
			}
			continue
		}
		if field.GetValidation() == nil {
			continue
		}
		validation := field.GetValidation()
		if text, ok := raw.(string); ok {
			if pattern := validation.GetPattern(); pattern != "" {
				matched, err := regexp.MatchString(pattern, text)
				if err != nil {
					return fmt.Errorf("connection config %q field %q has an invalid validation pattern", schema.GetKey(), field.GetKey())
				}
				if !matched {
					return fmt.Errorf("connection config %q field %q is invalid", schema.GetKey(), field.GetKey())
				}
			}
			length := utf8.RuneCountInString(text)
			if minimum := int(validation.GetMinLength()); minimum > 0 && length < minimum {
				return fmt.Errorf("connection config %q field %q must be at least %d characters", schema.GetKey(), field.GetKey(), minimum)
			}
			if maximum := int(validation.GetMaxLength()); maximum > 0 && length > maximum {
				return fmt.Errorf("connection config %q field %q must be at most %d characters", schema.GetKey(), field.GetKey(), maximum)
			}
		}
		if number, ok := connectionConfigNumber(raw); ok {
			if validation.GetHasMin() && number < validation.GetMin() {
				return fmt.Errorf("connection config %q field %q must be at least %g", schema.GetKey(), field.GetKey(), validation.GetMin())
			}
			if validation.GetHasMax() && number > validation.GetMax() {
				return fmt.Errorf("connection config %q field %q must be at most %g", schema.GetKey(), field.GetKey(), validation.GetMax())
			}
		}
	}
	return nil
}

func connectionConfigValueIsEmpty(value any) bool {
	switch typed := value.(type) {
	case nil:
		return true
	case string:
		return strings.TrimSpace(typed) == ""
	case []any:
		return len(typed) == 0
	default:
		return false
	}
}

func connectionAdminFieldIsVisible(
	form *pluginv1.AdminFormDescriptor,
	field *pluginv1.AdminFormField,
	values map[string]any,
) bool {
	if !connectionAdminConditionsMatch(field.GetShowWhen(), values, form.GetFields()) {
		return false
	}
	contained := false
	for _, section := range form.GetSections() {
		if section == nil || !stringSliceContains(section.GetFieldKeys(), field.GetKey()) {
			continue
		}
		contained = true
		if connectionAdminConditionsMatch(section.GetShowWhen(), values, form.GetFields()) {
			return true
		}
	}
	return !contained
}

func connectionAdminConditionsMatch(
	conditions []*pluginv1.AdminFormCondition,
	values map[string]any,
	fields []*pluginv1.AdminFormField,
) bool {
	for _, condition := range conditions {
		if condition == nil {
			continue
		}
		value, exists := values[condition.GetField()]
		if !exists {
			for _, field := range fields {
				if field != nil && field.GetKey() == condition.GetField() && field.GetDefaultValue() != nil {
					value = field.GetDefaultValue().AsInterface()
					break
				}
			}
		}
		if !stringSliceContains(condition.GetEquals(), connectionAdminConditionString(value)) {
			return false
		}
	}
	return true
}

func connectionAdminConditionString(value any) string {
	switch typed := value.(type) {
	case nil:
		return ""
	case bool:
		return strconv.FormatBool(typed)
	case float64:
		return strconv.FormatFloat(typed, 'f', -1, 64)
	case string:
		return typed
	default:
		return fmt.Sprint(typed)
	}
}

func stringSliceContains(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func connectionConfigNumber(value any) (float64, bool) {
	switch typed := value.(type) {
	case float64:
		return typed, true
	case float32:
		return float64(typed), true
	case int:
		return float64(typed), true
	case int8:
		return float64(typed), true
	case int16:
		return float64(typed), true
	case int32:
		return float64(typed), true
	case int64:
		return float64(typed), true
	case uint:
		return float64(typed), true
	case uint8:
		return float64(typed), true
	case uint16:
		return float64(typed), true
	case uint32:
		return float64(typed), true
	case uint64:
		return float64(typed), true
	case json.Number:
		number, err := typed.Float64()
		return number, err == nil
	case string:
		number, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
		return number, err == nil
	default:
		return 0, false
	}
}

func connectionConfigString(value any) (string, error) {
	if text, ok := value.(string); ok {
		return text, nil
	}
	encoded, err := json.Marshal(value)
	if err != nil {
		return "", err
	}
	return string(encoded), nil
}

func mergeWatchSyncProviderConfig(base, connection *pluginv1.WatchSyncProviderConfig) *pluginv1.WatchSyncProviderConfig {
	merged := &pluginv1.WatchSyncProviderConfig{Values: map[string]string{}, SecretValues: map[string]string{}}
	if base != nil {
		for key, value := range base.GetValues() {
			merged.Values[key] = value
		}
		for key, value := range base.GetSecretValues() {
			merged.SecretValues[key] = value
		}
	}
	if connection != nil {
		for key, value := range connection.GetValues() {
			delete(merged.SecretValues, key)
			merged.Values[key] = value
		}
		for key, value := range connection.GetSecretValues() {
			delete(merged.Values, key)
			merged.SecretValues[key] = value
		}
	}
	return merged
}

func (p *PluginProvider) persistUpdatedCredentials(
	ctx context.Context,
	conn Connection,
	credentials *pluginv1.WatchSyncCredentials,
) (Connection, error) {
	if p.repository == nil {
		return Connection{}, retryableProviderError{message: "watch sync plugin credential storage is unavailable"}
	}
	tokens, err := tokenSetFromProto(credentials)
	if err != nil {
		return Connection{}, err
	}
	conn = connectionWithTokens(conn, tokens)
	persisted, err := p.repository.UpsertConnection(ctx, conn)
	if err != nil {
		return Connection{}, retryableProviderError{message: "watch sync plugin credential update could not be persisted"}
	}
	return persisted, nil
}

func watchEventFromLocalPlay(play LocalPlay, origin pluginv1.WatchSyncOrigin) *pluginv1.WatchSyncEvent {
	return &pluginv1.WatchSyncEvent{
		EventId:         play.HistoryID,
		Operation:       pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_MARK_WATCHED,
		Origin:          origin,
		OccurredAt:      timestamppb.New(play.WatchedAt),
		WatchHistoryId:  play.HistoryID,
		DurationSeconds: play.DurationSeconds,
		ProviderItemKey: play.ProviderItemKey,
		Media: mediaFromIdentity(play.MediaItemID, play.Kind, play.Title, play.Year,
			play.IMDbID, play.TMDBID, play.TVDBID, play.SeriesTitle, play.SeriesYear,
			play.SeriesIMDbID, play.SeriesTMDBID, play.SeriesTVDBID, play.SeasonNumber, play.EpisodeNumber),
	}
}

func watchEventFromScrobble(event ScrobbleEvent, operation pluginv1.WatchSyncOperation) *pluginv1.WatchSyncEvent {
	eventID := "scrobble:" + operation.String() + ":" + event.PlaybackSessionID
	if operation == pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_PAUSE {
		eventID += fmt.Sprintf(":%.3f", event.PositionSeconds)
	}
	completion := 0.0
	if event.DurationSeconds > 0 {
		completion = event.PositionSeconds / event.DurationSeconds * 100
	}
	return &pluginv1.WatchSyncEvent{
		EventId:           eventID,
		Operation:         operation,
		Origin:            pluginv1.WatchSyncOrigin_WATCH_SYNC_ORIGIN_PLAYBACK_COMPLETION,
		OccurredAt:        timestamppb.New(event.OccurredAt),
		WatchHistoryId:    event.HistoryID,
		PlaybackSessionId: event.PlaybackSessionID,
		PositionSeconds:   event.PositionSeconds,
		DurationSeconds:   event.DurationSeconds,
		CompletionPercent: completion,
		Completed:         event.Completed,
		ProviderItemKey:   event.ProviderItemKey,
		Media: mediaFromIdentity(event.MediaItemID, event.Kind, "", 0,
			event.IMDbID, event.TMDBID, event.TVDBID, "", 0,
			event.SeriesIMDbID, event.SeriesTMDBID, event.SeriesTVDBID, event.SeasonNumber, event.EpisodeNumber),
	}
}

func mediaFromIdentity(mediaItemID, kind, title string, year int, imdbID, tmdbID, tvdbID, seriesTitle string, seriesYear int, seriesIMDbID, seriesTMDBID, seriesTVDBID string, season, episode int) *pluginv1.WatchSyncMedia {
	return &pluginv1.WatchSyncMedia{
		MediaItemId:       mediaItemID,
		MediaType:         watchSyncMediaType(kind),
		Title:             title,
		Year:              int32(year),
		ExternalIds:       watchIDs(imdbID, tmdbID, tvdbID),
		SeriesTitle:       seriesTitle,
		SeriesYear:        int32(seriesYear),
		SeriesExternalIds: watchIDs(seriesIMDbID, seriesTMDBID, seriesTVDBID),
		SeasonNumber:      int32(season),
		EpisodeNumber:     int32(episode),
	}
}

func watchSyncMediaType(kind string) pluginv1.WatchSyncMediaType {
	switch strings.ToLower(strings.TrimSpace(kind)) {
	case historyimport.KindMovie:
		return pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE
	case historyimport.KindEpisode:
		return pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_EPISODE
	default:
		return pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_UNSPECIFIED
	}
}

func watchIDs(imdbID, tmdbID, tvdbID string) map[string]string {
	ids := map[string]string{}
	if imdbID != "" {
		ids["imdb"] = imdbID
	}
	if tmdbID != "" {
		ids["tmdb"] = tmdbID
	}
	if tvdbID != "" {
		ids["tvdb"] = tvdbID
	}
	return ids
}

func credentialsFromConnection(conn Connection) *pluginv1.WatchSyncCredentials {
	credentials := &pluginv1.WatchSyncCredentials{
		AccessToken:      conn.AccessToken,
		RefreshToken:     conn.RefreshToken,
		TokenType:        conn.TokenType,
		Scopes:           append([]string(nil), conn.Scopes...),
		SecretAttributes: cloneStringMap(conn.SecretAttributes),
	}
	if conn.TokenExpiresAt != nil {
		credentials.ExpiresAt = timestamppb.New(*conn.TokenExpiresAt)
	}
	return credentials
}

func tokenSetFromProto(credentials *pluginv1.WatchSyncCredentials) (TokenSet, error) {
	if credentials == nil || strings.TrimSpace(credentials.GetAccessToken()) == "" {
		return TokenSet{}, errors.New("watch sync plugin returned no access token")
	}
	var expiresAt *time.Time
	if credentials.GetExpiresAt() != nil {
		if err := credentials.GetExpiresAt().CheckValid(); err != nil {
			return TokenSet{}, errors.New("watch sync plugin returned an invalid credential expiry")
		}
		value := credentials.GetExpiresAt().AsTime()
		expiresAt = &value
	}
	return TokenSet{
		AccessToken:      credentials.GetAccessToken(),
		RefreshToken:     credentials.GetRefreshToken(),
		TokenExpiresAt:   expiresAt,
		TokenType:        strings.TrimSpace(credentials.GetTokenType()),
		Scopes:           append([]string(nil), credentials.GetScopes()...),
		SecretAttributes: cloneStringMap(credentials.GetSecretAttributes()),
	}, nil
}

func accountFromProto(account *pluginv1.WatchSyncAccount) (ProviderAccount, error) {
	if account == nil || strings.TrimSpace(account.GetExternalSubject()) == "" {
		return ProviderAccount{}, errors.New("watch sync plugin returned no provider account identity")
	}
	return ProviderAccount{ID: account.GetExternalSubject(), Username: account.GetUsername()}, nil
}

func resultForEvent(results []*pluginv1.WatchSyncApplyResult, eventID string) *pluginv1.WatchSyncApplyResult {
	for _, result := range results {
		if result.GetEventId() == eventID {
			return result
		}
	}
	return nil
}

func supportedWatchSyncAuthMethod(descriptor *pluginv1.WatchSyncProviderDescriptor) (string, error) {
	supported := make(map[string]struct{}, 2)
	for _, method := range descriptor.GetAuthMethods() {
		switch method {
		case pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY:
			supported[AuthMethodAPIKey] = struct{}{}
		case pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE:
			supported[AuthMethodDeviceCode] = struct{}{}
		}
	}
	if len(supported) > 1 {
		return "", errors.New("advertises multiple host authentication methods; exactly one is required")
	}
	if _, ok := supported[AuthMethodAPIKey]; ok {
		return AuthMethodAPIKey, nil
	}
	if _, ok := supported[AuthMethodDeviceCode]; ok {
		return AuthMethodDeviceCode, nil
	}
	return "", errors.New("does not advertise an authentication method supported by the host")
}

func supportedWatchSyncMediaTypes(descriptor *pluginv1.WatchSyncProviderDescriptor) (map[pluginv1.WatchSyncMediaType]struct{}, error) {
	media := descriptor.GetSupportedMediaTypes()
	if len(media) == 0 {
		return map[pluginv1.WatchSyncMediaType]struct{}{
			pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE:   {},
			pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_EPISODE: {},
		}, nil
	}
	supported := make(map[pluginv1.WatchSyncMediaType]struct{}, len(media))
	for _, mediaType := range media {
		switch mediaType {
		case pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE,
			pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_EPISODE:
			supported[mediaType] = struct{}{}
		default:
			return nil, fmt.Errorf("advertises unsupported media type %q", mediaType.String())
		}
	}
	return supported, nil
}

func (p *PluginProvider) supportsMedia(mediaType pluginv1.WatchSyncMediaType) bool {
	if mediaType == pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_UNSPECIFIED {
		return false
	}
	_, ok := p.supportedMedia[mediaType]
	return ok
}

func unsupportedWatchSyncMediaMessage(mediaType pluginv1.WatchSyncMediaType) string {
	switch mediaType {
	case pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE:
		return watchSyncUnsupportedMovieMediaMessage
	case pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_EPISODE:
		return watchSyncUnsupportedEpisodeMediaMessage
	default:
		return watchSyncUnsupportedMediaMessage
	}
}

func safeApplyMessage(result *pluginv1.WatchSyncApplyResult, secrets ...string) string {
	if result == nil {
		return "watch sync provider could not apply the event"
	}
	return sanitizeWatchSyncMessage(result.GetFault().GetSafeMessage(), "watch sync provider could not apply the event", secrets...)
}

func sanitizeWatchSyncMessage(message string, fallback string, secrets ...string) string {
	message = normalizeWatchSyncText(message)
	for _, secret := range secrets {
		if secret = normalizeWatchSyncText(secret); secret != "" {
			message = strings.ReplaceAll(message, secret, "[REDACTED]")
		}
	}
	if message == "" {
		return fallback
	}
	const maxRunes = 256
	runes := []rune(message)
	if len(runes) > maxRunes {
		message = string(runes[:maxRunes]) + "…"
	}
	return message
}

func normalizeWatchSyncText(value string) string {
	value = strings.Map(func(r rune) rune {
		if unicode.IsControl(r) {
			return ' '
		}
		return r
	}, value)
	return strings.Join(strings.Fields(value), " ")
}

func watchSyncUnavailableError() error {
	return retryableProviderError{message: "watch sync plugin is unavailable"}
}

func watchSyncRPCError() error {
	return retryableProviderError{message: "watch sync plugin RPC failed"}
}

type watchSyncProviderFaultError struct {
	code    pluginv1.WatchSyncFaultCode
	message string
}

func (e watchSyncProviderFaultError) Error() string { return e.message }

func isWatchSyncInvalidCredentialError(err error) bool {
	var fault watchSyncProviderFaultError
	return errors.As(err, &fault) && fault.code == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_INVALID_CREDENTIAL
}

func watchSyncFaultError(provider string, fault *pluginv1.WatchSyncFault, secrets ...string) error {
	if fault == nil || fault.GetCode() == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_UNSPECIFIED {
		return nil
	}
	message := sanitizeWatchSyncMessage(fault.GetSafeMessage(), "watch sync provider request failed", secrets...)
	if fault.GetCode() == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_RATE_LIMITED {
		retry := time.Duration(0)
		if fault.GetRetryAfter() != nil {
			retry = fault.GetRetryAfter().AsDuration()
		}
		return RateLimitedError{Provider: provider, RetryAfter: retry}
	}
	if fault.GetCode() == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_TEMPORARY {
		return retryableProviderError{message: message}
	}
	return watchSyncProviderFaultError{code: fault.GetCode(), message: message}
}

func firstNonEmptyWatchID(values ...string) string {
	for _, value := range values {
		if value = strings.TrimSpace(value); value != "" {
			return value
		}
	}
	return "unknown"
}