silo-server/internal/watchsync/plugin_provider_state.go

package watchsync

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

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

const (
	pluginWatchedCursorKey   = "plugin.remote.watched"
	pluginProgressCursorKey  = "plugin.remote.progress"
	pluginFavoritesCursorKey = "plugin.remote.favorites"
	pluginWatchlistCursorKey = "plugin.remote.watchlist"
	maxRemoteStatePages      = 10_000
	maxRemoteStateItems      = 100_000
)

type pluginRemoteTraversal struct {
	items            []*pluginv1.WatchSyncRemoteState
	nextCursor       string
	completeSnapshot bool
	warnings         []string
}

func (p *PluginProvider) FetchWatched(
	ctx context.Context,
	cfg ServerConfig,
	conn Connection,
) ([]RemoteWatch, error) {
	batch, err := p.FetchWatchedBatch(ctx, cfg, conn)
	return batch.Rows, err
}

func (p *PluginProvider) FetchWatchedBatch(
	ctx context.Context,
	_ ServerConfig,
	conn Connection,
) (WatchedImportBatch, error) {
	traversal, err := p.listRemoteState(ctx, conn, pluginWatchedCursorKey,
		pluginv1.WatchSyncRemoteStateKind_WATCH_SYNC_REMOTE_STATE_KIND_WATCHED)
	if err != nil {
		return WatchedImportBatch{}, err
	}
	batch := WatchedImportBatch{
		UpdatedCursors: cursorUpdate(pluginWatchedCursorKey, traversal.nextCursor),
		Warnings:       traversal.warnings,
	}
	for _, state := range traversal.items {
		if state.GetWatched() == nil {
			continue
		}
		row, err := remoteWatchFromProto(p.Key(), state)
		if err != nil {
			batch.Warnings = append(batch.Warnings, err.Error())
			continue
		}
		batch.Rows = append(batch.Rows, row)
	}
	return batch, nil
}

func (p *PluginProvider) FetchProgress(
	ctx context.Context,
	cfg ServerConfig,
	conn Connection,
) ([]RemoteProgress, error) {
	batch, err := p.FetchProgressBatch(ctx, cfg, conn)
	return batch.Rows, err
}

func (p *PluginProvider) FetchProgressBatch(
	ctx context.Context,
	_ ServerConfig,
	conn Connection,
) (ProgressImportBatch, error) {
	traversal, err := p.listRemoteState(ctx, conn, pluginProgressCursorKey,
		pluginv1.WatchSyncRemoteStateKind_WATCH_SYNC_REMOTE_STATE_KIND_PROGRESS)
	if err != nil {
		return ProgressImportBatch{}, err
	}
	batch := ProgressImportBatch{
		UpdatedCursors: cursorUpdate(pluginProgressCursorKey, traversal.nextCursor),
		Warnings:       traversal.warnings,
	}
	for _, state := range traversal.items {
		if state.GetProgress() == nil {
			continue
		}
		row, err := remoteProgressFromProto(p.Key(), state)
		if err != nil {
			batch.Warnings = append(batch.Warnings, err.Error())
			continue
		}
		batch.Rows = append(batch.Rows, row)
	}
	return batch, nil
}

func (p *PluginProvider) FetchFavorites(
	ctx context.Context,
	cfg ServerConfig,
	conn Connection,
) ([]RemoteFavorite, error) {
	batch, err := p.FetchFavoritesBatch(ctx, cfg, conn)
	return batch.Rows, err
}

func (p *PluginProvider) FetchFavoritesBatch(
	ctx context.Context,
	_ ServerConfig,
	conn Connection,
) (FavoriteImportBatch, error) {
	return p.fetchListState(ctx, conn, pluginFavoritesCursorKey,
		pluginv1.WatchSyncRemoteStateKind_WATCH_SYNC_REMOTE_STATE_KIND_FAVORITE)
}

func (p *PluginProvider) FetchWatchlist(
	ctx context.Context,
	cfg ServerConfig,
	conn Connection,
) ([]RemoteFavorite, error) {
	batch, err := p.FetchWatchlistBatch(ctx, cfg, conn)
	return batch.Rows, err
}

func (p *PluginProvider) FetchWatchlistBatch(
	ctx context.Context,
	_ ServerConfig,
	conn Connection,
) (FavoriteImportBatch, error) {
	return p.fetchListState(ctx, conn, pluginWatchlistCursorKey,
		pluginv1.WatchSyncRemoteStateKind_WATCH_SYNC_REMOTE_STATE_KIND_WATCHLIST)
}

func (p *PluginProvider) fetchListState(
	ctx context.Context,
	conn Connection,
	cursorKey string,
	kind pluginv1.WatchSyncRemoteStateKind,
) (FavoriteImportBatch, error) {
	traversal, err := p.listRemoteState(ctx, conn, cursorKey, kind)
	if err != nil {
		return FavoriteImportBatch{}, err
	}
	batch := FavoriteImportBatch{
		UpdatedCursors: cursorUpdate(cursorKey, traversal.nextCursor),
		Warnings:       traversal.warnings,
		Incremental:    !traversal.completeSnapshot,
	}
	for _, state := range traversal.items {
		var listed *pluginv1.WatchSyncRemoteListState
		if kind == pluginv1.WatchSyncRemoteStateKind_WATCH_SYNC_REMOTE_STATE_KIND_FAVORITE {
			listed = state.GetFavorite()
		} else {
			listed = state.GetWatchlist()
		}
		if listed == nil {
			continue
		}
		row, err := remoteFavoriteFromProto(p.Key(), state, listed)
		if err != nil {
			batch.Warnings = append(batch.Warnings, err.Error())
			continue
		}
		batch.Rows = append(batch.Rows, row)
	}
	return batch, nil
}

func (p *PluginProvider) listRemoteState(
	ctx context.Context,
	conn Connection,
	cursorKey string,
	kind pluginv1.WatchSyncRemoteStateKind,
) (pluginRemoteTraversal, error) {
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return pluginRemoteTraversal{}, watchSyncUnavailableError()
	}
	cursor := conn.SyncCursors[cursorKey]
	pageToken := ""
	seenPageTokens := make(map[string]struct{})
	result := pluginRemoteTraversal{}
	snapshotSet := false
	for page := 0; page < maxRemoteStatePages; page++ {
		authContext, err := p.authenticatedContext(ctx, conn)
		if err != nil {
			return pluginRemoteTraversal{}, err
		}
		response, err := client.ListRemoteState(ctx, &pluginv1.WatchSyncListRemoteStateRequest{
			Context:    authContext,
			Cursor:     cursor,
			PageToken:  pageToken,
			PageSize:   int32(max(1, p.ExportBatchSize())),
			StateKinds: []pluginv1.WatchSyncRemoteStateKind{kind},
		})
		if err != nil {
			return pluginRemoteTraversal{}, watchSyncRPCError()
		}
		if response.GetUpdatedCredentials() != nil {
			conn, err = p.persistUpdatedCredentials(ctx, conn, response.GetUpdatedCredentials())
			if err != nil {
				return pluginRemoteTraversal{}, err
			}
		}
		if err := watchSyncFaultError(p.Key(), response.GetFault(), conn.AccessToken, conn.RefreshToken); err != nil {
			return pluginRemoteTraversal{}, err
		}
		if kind == pluginv1.WatchSyncRemoteStateKind_WATCH_SYNC_REMOTE_STATE_KIND_WATCHLIST &&
			p.descriptor.GetProvidesWatchlistOrder() && !response.GetCompleteSnapshot() {
			return pluginRemoteTraversal{}, errors.New("watch sync plugin returned an incremental traversal for an ordered watchlist")
		}
		if snapshotSet && result.completeSnapshot != response.GetCompleteSnapshot() {
			return pluginRemoteTraversal{}, errors.New("watch sync plugin changed snapshot mode during pagination")
		}
		result.completeSnapshot = response.GetCompleteSnapshot()
		snapshotSet = true
		items := response.GetItems()
		if len(items) > maxRemoteStateItems-len(result.items) {
			return pluginRemoteTraversal{}, errors.New("watch sync plugin exceeded the remote-state item limit")
		}
		result.items = append(result.items, items...)

		nextPage := strings.TrimSpace(response.GetNextPageToken())
		if nextPage == "" {
			result.nextCursor = response.GetNextCursor()
			return result, nil
		}
		if strings.TrimSpace(response.GetNextCursor()) != "" {
			return pluginRemoteTraversal{}, errors.New("watch sync plugin returned a durable cursor before the final page")
		}
		if _, duplicate := seenPageTokens[nextPage]; duplicate {
			return pluginRemoteTraversal{}, errors.New("watch sync plugin repeated a page token")
		}
		seenPageTokens[nextPage] = struct{}{}
		pageToken = nextPage
	}
	return pluginRemoteTraversal{}, errors.New("watch sync plugin exceeded the remote-state page limit")
}

func (p *PluginProvider) RemoveHistory(
	ctx context.Context,
	_ ServerConfig,
	conn Connection,
	plays []LocalPlay,
) (ExportResult, error) {
	result := ExportResult{Failed: make(map[string]string)}
	events := make([]*pluginv1.WatchSyncEvent, 0, len(plays))
	keys := make([]string, 0, len(plays))
	for _, play := range plays {
		event := watchEventFromLocalPlay(play, pluginv1.WatchSyncOrigin_WATCH_SYNC_ORIGIN_MANUAL)
		event.EventId = "unwatched:" + play.HistoryID
		event.Operation = pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_MARK_UNWATCHED
		event.ProviderItemKey = play.ProviderItemKey
		if !p.supportsMedia(event.GetMedia().GetMediaType()) {
			result.Failed[play.HistoryID] = unsupportedWatchSyncMediaMessage(event.GetMedia().GetMediaType())
			continue
		}
		events = append(events, event)
		keys = append(keys, play.HistoryID)
	}
	applied, err := p.applyPluginEvents(ctx, conn, events, keys)
	return mergeExportFailures(applied, result.Failed), err
}

func (p *PluginProvider) ExportFavorites(ctx context.Context, _ ServerConfig, conn Connection, items []LocalFavorite) (ExportResult, error) {
	return p.applyListEvents(ctx, conn, items, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_ADD_FAVORITE)
}

func (p *PluginProvider) RemoveFavorites(ctx context.Context, _ ServerConfig, conn Connection, items []LocalFavorite) (ExportResult, error) {
	return p.applyListEvents(ctx, conn, items, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_REMOVE_FAVORITE)
}

func (p *PluginProvider) ExportWatchlist(ctx context.Context, _ ServerConfig, conn Connection, items []LocalFavorite) (ExportResult, error) {
	return p.applyListEvents(ctx, conn, items, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_ADD_TO_WATCHLIST)
}

func (p *PluginProvider) RemoveWatchlist(ctx context.Context, _ ServerConfig, conn Connection, items []LocalFavorite) (ExportResult, error) {
	return p.applyListEvents(ctx, conn, items, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_REMOVE_FROM_WATCHLIST)
}

func (p *PluginProvider) applyListEvents(
	ctx context.Context,
	conn Connection,
	items []LocalFavorite,
	operation pluginv1.WatchSyncOperation,
) (ExportResult, error) {
	result := ExportResult{Failed: make(map[string]string)}
	events := make([]*pluginv1.WatchSyncEvent, 0, len(items))
	keys := make([]string, 0, len(items))
	for _, item := range items {
		media := mediaFromIdentity(item.MediaItemID, item.Kind, item.Title, item.Year,
			item.IMDbID, item.TMDBID, item.TVDBID, "", 0,
			item.SeriesIMDbID, item.SeriesTMDBID, item.SeriesTVDBID, 0, 0)
		if !p.supportsMedia(media.GetMediaType()) {
			result.Failed[item.MediaItemID] = unsupportedWatchSyncMediaMessage(media.GetMediaType())
			continue
		}
		event := &pluginv1.WatchSyncEvent{
			EventId:         fmt.Sprintf("%s:%s", operation.String(), item.MediaItemID),
			Operation:       operation,
			Origin:          pluginv1.WatchSyncOrigin_WATCH_SYNC_ORIGIN_MANUAL,
			OccurredAt:      timestampOrNil(item.FavoritedAt),
			Media:           media,
			ProviderItemKey: item.ProviderItemKey,
		}
		if operation == pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_ADD_TO_WATCHLIST {
			position := int32(len(events))
			event.ListPosition = &position
		}
		events = append(events, event)
		keys = append(keys, item.MediaItemID)
	}
	applied, err := p.applyPluginEvents(ctx, conn, events, keys)
	return mergeExportFailures(applied, result.Failed), err
}

func (p *PluginProvider) applyPluginEvents(
	ctx context.Context,
	conn Connection,
	events []*pluginv1.WatchSyncEvent,
	keys []string,
) (ExportResult, error) {
	result, _, err := p.applyPluginEventsDetailed(ctx, conn, events, keys)
	return result, err
}

func (p *PluginProvider) applyPluginEventsDetailed(
	ctx context.Context,
	conn Connection,
	events []*pluginv1.WatchSyncEvent,
	keys []string,
) (ExportResult, map[string]pluginv1.WatchSyncApplyStatus, error) {
	result := ExportResult{Failed: make(map[string]string)}
	statuses := make(map[string]pluginv1.WatchSyncApplyStatus, len(events))
	if len(events) == 0 {
		return result, statuses, nil
	}
	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
	if err != nil {
		return result, statuses, watchSyncUnavailableError()
	}
	for offset := 0; offset < len(events); offset += max(1, p.ExportBatchSize()) {
		end := min(len(events), offset+max(1, p.ExportBatchSize()))
		authContext, err := p.authenticatedContext(ctx, conn)
		if err != nil {
			return result, statuses, err
		}
		response, err := client.ApplyEvents(ctx, &pluginv1.WatchSyncApplyEventsRequest{
			Context: authContext,
			Events:  events[offset:end],
		})
		if err != nil {
			return result, statuses, watchSyncRPCError()
		}
		if response.GetUpdatedCredentials() != nil {
			conn, err = p.persistUpdatedCredentials(ctx, conn, response.GetUpdatedCredentials())
			if err != nil {
				return result, statuses, err
			}
		}
		if err := watchSyncFaultError(p.Key(), response.GetFault(), conn.AccessToken, conn.RefreshToken); err != nil {
			return result, statuses, err
		}
		for index, event := range events[offset:end] {
			key := keys[offset+index]
			apply := resultForEvent(response.GetResults(), event.GetEventId())
			statuses[key] = apply.GetStatus()
			switch apply.GetStatus() {
			case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_APPLIED,
				pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_NO_CHANGE:
				result.Sent = append(result.Sent, key)
			case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_REJECTED:
				result.NotFound = append(result.NotFound, key)
			case pluginv1.WatchSyncApplyStatus_WATCH_SYNC_APPLY_STATUS_RETRY:
				fault := apply.GetFault()
				if fault.GetCode() == pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_RATE_LIMITED {
					retry := time.Duration(0)
					if fault.GetRetryAfter() != nil {
						retry = fault.GetRetryAfter().AsDuration()
					}
					return result, statuses, RateLimitedError{Provider: p.Key(), RetryAfter: retry}
				}
				result.Failed[key] = safeApplyMessage(apply, conn.AccessToken, conn.RefreshToken)
			default:
				result.Failed[key] = "watch sync plugin omitted a valid event result"
			}
		}
	}
	return result, statuses, nil
}

func mergeExportFailures(result ExportResult, failures map[string]string) ExportResult {
	if result.Failed == nil {
		result.Failed = make(map[string]string, len(failures))
	}
	for key, message := range failures {
		result.Failed[key] = message
	}
	return result
}

func remoteWatchFromProto(provider string, state *pluginv1.WatchSyncRemoteState) (RemoteWatch, error) {
	identity, err := remoteIdentityFromProto(state)
	if err != nil {
		return RemoteWatch{}, err
	}
	watched := state.GetWatched()
	if watched.GetPlayCount() < 1 {
		return RemoteWatch{}, errors.New("watch sync plugin returned watched state with no plays")
	}
	return RemoteWatch{
		Provider: provider, ProviderItemKey: state.GetProviderItemKey(),
		Kind: identity.kind, Title: identity.title, Year: identity.year,
		IMDbID: identity.imdbID, TMDBID: identity.tmdbID, TVDBID: identity.tvdbID,
		SeriesTitle: identity.seriesTitle, SeriesYear: identity.seriesYear,
		SeriesIMDbID: identity.seriesIMDbID, SeriesTMDBID: identity.seriesTMDBID, SeriesTVDBID: identity.seriesTVDBID,
		SeasonNumber: identity.season, EpisodeNumber: identity.episode,
		PlayCount: int(watched.GetPlayCount()), LastWatchedAt: timePointer(watched.GetLastWatchedAt()),
	}, nil
}

func remoteProgressFromProto(provider string, state *pluginv1.WatchSyncRemoteState) (RemoteProgress, error) {
	identity, err := remoteIdentityFromProto(state)
	if err != nil {
		return RemoteProgress{}, err
	}
	progress := state.GetProgress()
	if progress.GetProgressPercent() < 0 || progress.GetProgressPercent() >= 100 {
		return RemoteProgress{}, errors.New("watch sync plugin returned progress outside [0,100)")
	}
	pausedAt := timePointer(progress.GetPausedAt())
	if pausedAt == nil {
		return RemoteProgress{}, errors.New("watch sync plugin returned progress without a valid timestamp")
	}
	return RemoteProgress{
		Provider: provider, ProviderItemKey: state.GetProviderItemKey(),
		Kind: identity.kind, Title: identity.title, Year: identity.year,
		IMDbID: identity.imdbID, TMDBID: identity.tmdbID, TVDBID: identity.tvdbID,
		SeriesTitle: identity.seriesTitle, SeriesYear: identity.seriesYear,
		SeriesIMDbID: identity.seriesIMDbID, SeriesTMDBID: identity.seriesTMDBID, SeriesTVDBID: identity.seriesTVDBID,
		SeasonNumber: identity.season, EpisodeNumber: identity.episode,
		ProgressPercent: progress.GetProgressPercent(), PausedAt: *pausedAt,
	}, nil
}

func remoteFavoriteFromProto(provider string, state *pluginv1.WatchSyncRemoteState, listed *pluginv1.WatchSyncRemoteListState) (RemoteFavorite, error) {
	if state == nil || listed == nil {
		return RemoteFavorite{}, errors.New("watch sync plugin returned list state without identity")
	}
	providerItemKey := strings.TrimSpace(state.GetProviderItemKey())
	if listed.GetRemoved() {
		if providerItemKey == "" {
			return RemoteFavorite{}, errors.New("watch sync plugin returned a list tombstone without provider identity")
		}
		return RemoteFavorite{
			Provider:        provider,
			ProviderItemKey: providerItemKey,
			Removed:         true,
		}, nil
	}
	identity, err := remoteIdentityFromProto(state)
	if err != nil {
		return RemoteFavorite{}, err
	}
	listedAt := time.Now().UTC()
	if value := timePointer(listed.GetListedAt()); value != nil {
		listedAt = *value
	}
	return RemoteFavorite{
		Provider: provider, ProviderItemKey: providerItemKey,
		Kind: identity.kind, Title: identity.title, Year: identity.year,
		IMDbID: identity.imdbID, TMDBID: identity.tmdbID, TVDBID: identity.tvdbID,
		SeriesTitle: identity.seriesTitle, SeriesYear: identity.seriesYear,
		SeriesIMDbID: identity.seriesIMDbID, SeriesTMDBID: identity.seriesTMDBID, SeriesTVDBID: identity.seriesTVDBID,
		SeasonNumber: identity.season, EpisodeNumber: identity.episode, FavoritedAt: listedAt,
	}, nil
}

type remoteIdentity struct {
	kind, title, imdbID, tmdbID, tvdbID                   string
	seriesTitle, seriesIMDbID, seriesTMDBID, seriesTVDBID string
	year, seriesYear, season, episode                     int
}

func remoteIdentityFromProto(state *pluginv1.WatchSyncRemoteState) (remoteIdentity, error) {
	if state == nil || state.GetMedia() == nil || strings.TrimSpace(state.GetProviderItemKey()) == "" {
		return remoteIdentity{}, errors.New("watch sync plugin returned remote state without identity")
	}
	media := state.GetMedia()
	var kind string
	switch media.GetMediaType() {
	case pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_MOVIE:
		kind = historyimport.KindMovie
	case pluginv1.WatchSyncMediaType_WATCH_SYNC_MEDIA_TYPE_EPISODE:
		kind = historyimport.KindEpisode
	default:
		return remoteIdentity{}, errors.New("watch sync plugin returned unsupported remote media")
	}
	return remoteIdentity{
		kind: kind, title: media.GetTitle(), year: int(media.GetYear()),
		imdbID: media.GetExternalIds()["imdb"], tmdbID: media.GetExternalIds()["tmdb"], tvdbID: media.GetExternalIds()["tvdb"],
		seriesTitle: media.GetSeriesTitle(), seriesYear: int(media.GetSeriesYear()),
		seriesIMDbID: media.GetSeriesExternalIds()["imdb"], seriesTMDBID: media.GetSeriesExternalIds()["tmdb"], seriesTVDBID: media.GetSeriesExternalIds()["tvdb"],
		season: int(media.GetSeasonNumber()), episode: int(media.GetEpisodeNumber()),
	}, nil
}

func cursorUpdate(key, value string) map[string]string {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return map[string]string{key: value}
}

func timePointer(value interface {
	AsTime() time.Time
	CheckValid() error
}) *time.Time {
	if value == nil || value.CheckValid() != nil {
		return nil
	}
	result := value.AsTime()
	return &result
}

func timestampOrNil(value time.Time) *timestamppb.Timestamp {
	if value.IsZero() {
		return nil
	}
	return timestamppb.New(value)
}