silo-server/internal/markers/plugin_provider.go

package markers

import (
	"context"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"google.golang.org/genproto/googleapis/rpc/errdetails"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/structpb"

	pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
	"github.com/Silo-Server/silo-server/internal/models"
	"github.com/Silo-Server/silo-server/internal/pluginhost"
)

const pluginMarkerAlgorithm = "plugin:marker_provider.v1"

type pluginMarkerClient interface {
	FetchMarkers(ctx context.Context, req *pluginv1.FetchMarkersRequest) (*pluginv1.FetchMarkersResponse, error)
	SubmitMarker(ctx context.Context, req *pluginv1.SubmitMarkerRequest) (*pluginv1.SubmitMarkerResponse, error)
	GetMarkerProviderStats(ctx context.Context, req *pluginv1.GetMarkerProviderStatsRequest) (*pluginv1.MarkerProviderStatsResponse, error)
}

type pluginMarkerResolver interface {
	MarkerProviderClient(ctx context.Context, installationID int, capabilityID string) (pluginMarkerClient, error)
}

type pluginMarkerClientFactory func(ctx context.Context, installationID int, capabilityID string) (pluginMarkerClient, error)

type PluginResolverAdapter struct {
	inner interface {
		MarkerProviderClient(ctx context.Context, installationID int, capabilityID string) (*pluginhost.MarkerProviderClient, error)
	}
}

func NewPluginResolverAdapter(svc interface {
	MarkerProviderClient(ctx context.Context, installationID int, capabilityID string) (*pluginhost.MarkerProviderClient, error)
}) *PluginResolverAdapter {
	if svc == nil {
		return nil
	}
	return &PluginResolverAdapter{inner: svc}
}

func (a *PluginResolverAdapter) MarkerProviderClient(ctx context.Context, installationID int, capabilityID string) (pluginMarkerClient, error) {
	return a.inner.MarkerProviderClient(ctx, installationID, capabilityID)
}

type PluginProviderOptions struct {
	InstallationID      int
	CapabilityID        string
	DisplayName         string
	PluginID            string
	RequiredExternalIDs []string
}

// PluginProvider adapts a marker_provider.v1 plugin capability to Silo's
// marker Provider/Submitter interfaces.
type PluginProvider struct {
	installationID      int
	capabilityID        string
	displayName         string
	pluginID            string
	requiredExternalIDs []string
	clientFactory       pluginMarkerClientFactory
}

var _ Submitter = (*PluginProvider)(nil)

func NewPluginProvider(opts PluginProviderOptions, resolver pluginMarkerResolver) (*PluginProvider, error) {
	if resolver == nil {
		return nil, fmt.Errorf("plugin marker resolver is required")
	}
	return NewPluginProviderWithClientFactory(opts, resolver.MarkerProviderClient)
}

func NewPluginProviderWithClientFactory(opts PluginProviderOptions, clientFactory pluginMarkerClientFactory) (*PluginProvider, error) {
	if opts.InstallationID <= 0 {
		return nil, fmt.Errorf("plugin marker installation id is required")
	}
	if strings.TrimSpace(opts.CapabilityID) == "" {
		return nil, fmt.Errorf("plugin marker capability id is required")
	}
	if clientFactory == nil {
		return nil, fmt.Errorf("plugin marker client factory is required")
	}
	displayName := strings.TrimSpace(opts.DisplayName)
	if displayName == "" {
		displayName = opts.CapabilityID
	}
	requiredIDs := normalizeExternalIDKeys(opts.RequiredExternalIDs)
	return &PluginProvider{
		installationID:      opts.InstallationID,
		capabilityID:        strings.TrimSpace(opts.CapabilityID),
		displayName:         displayName,
		pluginID:            strings.TrimSpace(opts.PluginID),
		requiredExternalIDs: requiredIDs,
		clientFactory:       clientFactory,
	}, nil
}

func PluginProviderID(installationID int, capabilityID string) string {
	return fmt.Sprintf("plugin:%d:%s", installationID, strings.TrimSpace(capabilityID))
}

func (p *PluginProvider) ID() string {
	if p == nil {
		return ""
	}
	return PluginProviderID(p.installationID, p.capabilityID)
}

func (p *PluginProvider) ProviderDescription() ProviderDescriptor {
	if p == nil {
		return ProviderDescriptor{}
	}
	return ProviderDescriptor{
		ID:                   p.ID(),
		DisplayName:          p.displayName,
		SourceType:           ProviderSourcePlugin,
		PluginID:             p.pluginID,
		PluginInstallationID: p.installationID,
		CapabilityID:         p.capabilityID,
	}
}

func (p *PluginProvider) SubmissionRequirements() SubmissionRequirements {
	if p == nil || len(p.requiredExternalIDs) == 0 {
		return SubmissionRequirements{}
	}
	return SubmissionRequirements{RequiredExternalIDs: append([]string(nil), p.requiredExternalIDs...)}
}

func (p *PluginProvider) FetchMarkers(ctx context.Context, req Request) (Result, error) {
	if p == nil || p.clientFactory == nil {
		return Result{}, nil
	}
	client, err := p.clientFactory(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return Result{}, err
	}
	response, err := client.FetchMarkers(ctx, &pluginv1.FetchMarkersRequest{
		ItemType:        itemTypeName(req.Kind),
		ExternalIds:     externalIDsProto(req.ExternalIDs),
		SeasonNumber:    int32(req.SeasonNumber),
		EpisodeNumber:   int32(req.EpisodeNumber),
		DurationSeconds: req.Duration.Seconds(),
	})
	if err != nil {
		return Result{}, pluginProviderError(p.ID(), err)
	}
	if response == nil {
		return Result{}, nil
	}

	result := Result{
		SourceClass: models.MarkerSourcePlugin,
		ProviderID:  p.ID(),
		Algorithm:   pluginMarkerAlgorithm,
	}
	for _, segment := range response.GetMarkers() {
		marker, ok := markerFromPluginSegment(segment, req.Duration)
		if !ok {
			continue
		}
		marker.SourceClass = models.MarkerSourcePlugin
		marker.ProviderID = p.ID()
		if marker.Algorithm == "" {
			marker.Algorithm = pluginMarkerAlgorithm
		}
		result.Markers = append(result.Markers, marker)
	}
	return result, nil
}

func (p *PluginProvider) SubmitMarker(ctx context.Context, req SubmissionRequest) (SubmissionResult, error) {
	if p == nil || p.clientFactory == nil {
		return SubmissionResult{}, fmt.Errorf("plugin marker provider not configured")
	}
	client, err := p.clientFactory(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return SubmissionResult{}, err
	}
	body := &pluginv1.SubmitMarkerRequest{
		ItemType:        itemTypeName(req.Kind),
		ExternalIds:     externalIDsProto(req.ExternalIDs),
		SeasonNumber:    int32(req.SeasonNumber),
		EpisodeNumber:   int32(req.EpisodeNumber),
		Segment:         markerKindName(req.Segment),
		DurationSeconds: req.Duration.Seconds(),
	}
	if req.Start != nil {
		start := req.Start.Seconds()
		body.StartSeconds = &start
	}
	if req.End != nil {
		end := req.End.Seconds()
		body.EndSeconds = &end
	}
	response, err := client.SubmitMarker(ctx, body)
	if err != nil {
		return SubmissionResult{}, pluginProviderError(p.ID(), err)
	}
	if response == nil {
		return SubmissionResult{Status: SubmissionStatusPending}, nil
	}
	status := strings.TrimSpace(response.GetStatus())
	if status == "" {
		status = SubmissionStatusPending
	}
	return SubmissionResult{ID: response.GetSubmissionId(), Status: status, Weight: response.GetWeight()}, nil
}

func (p *PluginProvider) FetchUserStats(ctx context.Context) (UserStats, error) {
	if p == nil || p.clientFactory == nil {
		return UserStats{}, fmt.Errorf("plugin marker provider not configured")
	}
	client, err := p.clientFactory(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return UserStats{}, err
	}
	response, err := client.GetMarkerProviderStats(ctx, &pluginv1.GetMarkerProviderStatsRequest{})
	if err != nil {
		return UserStats{}, pluginProviderError(p.ID(), err)
	}
	if response == nil {
		return UserStats{}, nil
	}
	return UserStats{
		Total:          int(response.GetTotal()),
		Accepted:       int(response.GetAccepted()),
		Pending:        int(response.GetPending()),
		Rejected:       int(response.GetRejected()),
		AcceptanceRate: response.GetAcceptanceRate(),
		CurrentStreak:  int(response.GetCurrentStreak()),
		BestStreak:     int(response.GetBestStreak()),
	}, nil
}

func markerFromPluginSegment(segment *pluginv1.MarkerSegment, duration time.Duration) (Marker, bool) {
	if segment == nil {
		return Marker{}, false
	}
	kind, ok := markerKindFromName(segment.GetSegment())
	if !ok {
		return Marker{}, false
	}
	start := time.Duration(0)
	end := duration
	if segment.StartSeconds != nil {
		start = secondsDuration(segment.GetStartSeconds())
	}
	if segment.EndSeconds != nil {
		end = secondsDuration(segment.GetEndSeconds())
	}
	if start < 0 {
		return Marker{}, false
	}
	if duration > 0 && (start >= duration || end > duration) {
		return Marker{}, false
	}
	if end <= start {
		return Marker{}, false
	}
	return Marker{
		Kind:            kind,
		Start:           start,
		End:             end,
		Confidence:      segment.GetConfidence(),
		SubmissionCount: int(segment.GetSubmissionCount()),
		Algorithm:       strings.TrimSpace(segment.GetAlgorithm()),
	}, true
}

func externalIDsProto(ids map[string]string) *pluginv1.MarkerExternalIDs {
	if len(ids) == 0 {
		return &pluginv1.MarkerExternalIDs{}
	}
	providerIDs := make(map[string]any, len(ids))
	for key, value := range ids {
		if strings.TrimSpace(value) != "" {
			providerIDs[key] = value
		}
	}
	structIDs, _ := structpb.NewStruct(providerIDs)
	return &pluginv1.MarkerExternalIDs{
		TmdbId:      strings.TrimSpace(ids[ExternalIDKeyTMDB]),
		ImdbId:      strings.TrimSpace(ids[ExternalIDKeyIMDB]),
		TvdbId:      strings.TrimSpace(ids[ExternalIDKeyTVDB]),
		ProviderIds: structIDs,
	}
}

func itemTypeName(kind ItemKind) string {
	switch kind {
	case ItemKindEpisode:
		return "episode"
	case ItemKindMovie:
		return "movie"
	default:
		return ""
	}
}

func markerKindName(kind MarkerKind) string {
	switch kind {
	case MarkerKindIntro:
		return "intro"
	case MarkerKindCredits:
		return "credits"
	case MarkerKindRecap:
		return "recap"
	case MarkerKindPreview:
		return "preview"
	default:
		return ""
	}
}

func markerKindFromName(name string) (MarkerKind, bool) {
	switch strings.ToLower(strings.TrimSpace(name)) {
	case "intro":
		return MarkerKindIntro, true
	case "credits":
		return MarkerKindCredits, true
	case "recap":
		return MarkerKindRecap, true
	case "preview":
		return MarkerKindPreview, true
	default:
		return 0, false
	}
}

func secondsDuration(seconds float64) time.Duration {
	return time.Duration(seconds * float64(time.Second))
}

func normalizeExternalIDKeys(keys []string) []string {
	seen := map[string]struct{}{}
	out := make([]string, 0, len(keys))
	for _, key := range keys {
		key = strings.ToLower(strings.TrimSpace(key))
		if key == "" {
			continue
		}
		switch key {
		case ExternalIDKeyTMDB, ExternalIDKeyIMDB, ExternalIDKeyTVDB:
		default:
			continue
		}
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		out = append(out, key)
	}
	return out
}

func pluginProviderError(providerID string, err error) error {
	if err == nil {
		return nil
	}
	st, ok := status.FromError(err)
	message := err.Error()
	if ok {
		message = st.Message()
	}
	upstreamStatus, legacyHTTPConflict := pluginSubmissionHTTPStatus(message)
	if (ok && st.Code() == codes.AlreadyExists) || (legacyHTTPConflict && upstreamStatus == http.StatusConflict) {
		return &SubmissionConflictError{
			Provider:   providerID,
			HTTPStatus: http.StatusConflict,
			Message:    err.Error(),
		}
	}
	if !ok || st.Code() != codes.ResourceExhausted {
		return err
	}
	retryAfter := time.Duration(0)
	for _, detail := range st.Details() {
		if retry, ok := detail.(*errdetails.RetryInfo); ok && retry.GetRetryDelay() != nil {
			retryAfter = retry.GetRetryDelay().AsDuration()
			break
		}
	}
	return &RetryAfterError{Provider: providerID, RetryAfter: retryAfter, Message: err.Error()}
}

// pluginSubmissionHTTPStatus recognizes the legacy marker-provider error text
// used before plugins could expose an AlreadyExists gRPC status. Keep this
// narrow to submit errors so unrelated provider failures are not reclassified.
func pluginSubmissionHTTPStatus(message string) (int, bool) {
	const marker = "submit HTTP "
	start := strings.Index(message, marker)
	if start < 0 {
		return 0, false
	}
	tail := strings.TrimSpace(message[start+len(marker):])
	if end := strings.IndexAny(tail, ": "); end >= 0 {
		tail = tail[:end]
	}
	code, err := strconv.Atoi(tail)
	if err != nil || code < 100 || code > 599 {
		return 0, false
	}
	return code, true
}

func PluginRequiredExternalIDsFromMetadata(metadata map[string]any) []string {
	raw, ok := metadata["required_external_ids"]
	if !ok {
		return nil
	}
	switch typed := raw.(type) {
	case []string:
		return normalizeExternalIDKeys(typed)
	case []any:
		keys := make([]string, 0, len(typed))
		for _, value := range typed {
			if s, ok := value.(string); ok {
				keys = append(keys, s)
			}
		}
		return normalizeExternalIDKeys(keys)
	case string:
		if strings.TrimSpace(typed) == "" {
			return nil
		}
		return normalizeExternalIDKeys(strings.Split(typed, ","))
	default:
		return nil
	}
}

func PluginDefaultFetchPriorityFromMetadata(metadata map[string]any) (int, bool) {
	raw, ok := metadata["default_fetch_priority"]
	if !ok {
		return 0, false
	}
	switch typed := raw.(type) {
	case int:
		return typed, true
	case int32:
		return int(typed), true
	case int64:
		return int(typed), true
	case float64:
		return int(typed), true
	case string:
		parsed, err := strconv.Atoi(strings.TrimSpace(typed))
		return parsed, err == nil
	default:
		return 0, false
	}
}