feat(trakt): catalog API and shared catalog core (watchlist, discover, recommendations)

lib/services/catalog/catalog_source.dart

@@ -0,0 +1,109 @@
+import 'package:flutter/foundation.dart';
+
+import '../../media/media_kind.dart';
+import '../../models/catalog/catalog_cast_member.dart';
+import '../../models/catalog/catalog_item.dart';
+import '../../utils/external_ids.dart';
+
+/// Content rows a catalog source can serve on the Explore tab.
+enum CatalogRowId {
+  watchlist,
+  recommendedMovies,
+  recommendedShows,
+  trendingMovies,
+  trendingShows,
+  popularMovies,
+  popularShows,
+  // Anime rows (MAL has no movie/show split).
+  suggestedAnime,
+  airingAnime,
+  popularAnime,
+  // Seerr rows (its trending endpoint is mixed movie/TV).
+  trending,
+  upcomingMovies,
+  upcomingShows,
+}
+
+/// Notify-guarded [ChangeNotifier] for [CatalogSource.watchlistChanges]: a
+/// snapshot load or mutation that resolves after the source was disposed
+/// (provider disconnected mid-session) must not trip the used-after-dispose
+/// assert.
+class WatchlistChangeNotifier extends ChangeNotifier {
+  bool _disposed = false;
+
+  void notify() {
+    if (!_disposed) notifyListeners();
+  }
+
+  @override
+  void dispose() {
+    _disposed = true;
+    super.dispose();
+  }
+}
+
+/// One page of a catalog row.
+class CatalogPage {
+  final List<CatalogItem> items;
+  final bool hasMore;
+
+  const CatalogPage({required this.items, this.hasMore = false});
+}
+
+/// A pluggable external catalog provider backing the Explore tab (Trakt
+/// today; Overseerr/Jellyfin or MAL later).
+///
+/// Implementations wrap an authenticated API client owned by their account
+/// provider; disposing a source must not dispose that client.
+abstract class CatalogSource {
+  CatalogSourceId get id;
+
+  String get displayName;
+
+  /// Rows this source serves, in display order.
+  List<CatalogRowId> get supportedRows;
+
+  /// Whether the source has a user watchlist that can be read and mutated.
+  bool get supportsWatchlist;
+
+  Future<CatalogPage> fetchRow(CatalogRowId row, {int page = 1, int limit = 25});
+
+  /// Free-text title search for the Explore search screen. Returns an empty
+  /// list when the query is below the provider's minimum length (MAL
+  /// rejects queries under 3 characters).
+  Future<List<CatalogItem>> search(String query, {int limit = 30});
+
+  /// Cast of an item for its detail screen (actors with characters, or MAL
+  /// characters with roles), in billing order. One request, fetched lazily
+  /// on detail open; empty when the provider has none for this item.
+  Future<List<CatalogCastMember>> fetchCast(CatalogItem item, {int limit = 20});
+
+  /// "More like this" titles for an item's detail screen (Trakt related,
+  /// MAL recommendations, Seerr/TMDB recommendations). One request, fetched
+  /// lazily on detail open; empty when the provider has none.
+  Future<List<CatalogItem>> fetchRelated(CatalogItem item, {int limit = 20});
+
+  /// Load the full watchlist membership snapshot (coalesced; cached for the
+  /// session). [isOnWatchlist] returns null until this has completed once.
+  Future<void> ensureWatchlistLoaded();
+
+  /// Whether the item is on the user's watchlist, or null when the snapshot
+  /// has not loaded yet.
+  bool? isOnWatchlist(MediaKind kind, CatalogItemIds ids);
+
+  /// Resolve the ids this source needs for watchlist membership/mutation of
+  /// a library item, given the external ids its server knows. Returns null
+  /// when the item cannot exist in this source's domain (e.g. non-anime for
+  /// MAL) — callers hide the watchlist action then.
+  Future<CatalogItemIds?> resolveItemIds(MediaKind kind, ExternalIds external);
+
+  Future<void> addToWatchlist(MediaKind kind, CatalogItemIds ids);
+
+  Future<void> removeFromWatchlist(MediaKind kind, CatalogItemIds ids);
+
+  /// Fires after any watchlist membership change (mutation or snapshot load)
+  /// so watchers (Explore rows, detail-screen buttons) can rebuild.
+  Listenable get watchlistChanges;
+
+  void dispose();
+}

lib/services/catalog/catalog_watchlist_machinery.dart

@@ -0,0 +1,165 @@
+import 'package:flutter/foundation.dart';
+
+import '../../media/media_kind.dart';
+import '../../models/catalog/catalog_item.dart';
+import '../../utils/app_logger.dart';
+import '../trackers/future_coalescer.dart';
+import 'catalog_source.dart';
+
+/// One snapshot page of watchlist membership: each group holds every key
+/// form of a single watchlist entry.
+typedef WatchlistKeyPage = ({List<List<String>> groups, bool hasMore});
+
+/// Shared watchlist snapshot + optimistic-mutation machinery behind the
+/// watchlist half of [CatalogSource] (Trakt, MAL). Implementations supply
+/// key derivation, the snapshot page fetch, and the actual mutation call.
+///
+/// The snapshot maps every key form of an entry to the entry's full key
+/// group, so a mutation carrying only a subset of the entry's id forms (a
+/// media-detail remove works from server-resolved external ids, which lack
+/// Trakt's trakt/slug forms) still drops the WHOLE entry. With a flat key
+/// set, the sibling keys survived a remove and any-match membership stayed
+/// true for the rest of the session.
+mixin CatalogWatchlistMachinery {
+  final WatchlistChangeNotifier _watchlistChanges = WatchlistChangeNotifier();
+  final FutureCoalescer<void> _watchlistLoad = FutureCoalescer();
+  Map<String, Set<String>>? _watchlistKeyGroups;
+
+  // ---------- Contract ----------
+
+  /// Log prefix naming the source and its list, e.g. `Trakt: watchlist`.
+  String get watchlistLogLabel;
+
+  /// Full-snapshot paging bounds.
+  int get watchlistPageLimit;
+  int get watchlistMaxPages;
+
+  /// Every membership key form of [ids] under [kind]'s namespace rules.
+  /// Empty when [ids] carries no form this source can key on.
+  List<String> membershipKeysFor(MediaKind kind, CatalogItemIds ids);
+
+  /// One page of the snapshot as key groups (one group per entry).
+  Future<WatchlistKeyPage> fetchWatchlistKeyPage(int page, int limit);
+
+  /// Resolve [ids] to the concrete forms the mutation call needs (MAL maps
+  /// external ids to a MAL id via Fribb). Throw when the item cannot exist
+  /// in this source's domain. Default: pass-through.
+  Future<CatalogItemIds> resolveWatchlistMutationIds(MediaKind kind, CatalogItemIds ids) async => ids;
+
+  /// The actual API mutation for the resolved [ids].
+  Future<void> performWatchlistMutation(MediaKind kind, CatalogItemIds ids, {required bool add});
+
+  // ---------- CatalogSource watchlist surface ----------
+
+  Listenable get watchlistChanges => _watchlistChanges;
+
+  /// Load failures are logged and swallowed: membership stays unknown
+  /// (null) and the next call retries — every UI call site fires this
+  /// unawaited, so a flaky request must not become an uncaught error.
+  Future<void> ensureWatchlistLoaded() {
+    if (_watchlistKeyGroups != null) return Future.value();
+    return _watchlistLoad.run(_loadWatchlistSnapshot);
+  }
+
+  Future<void> _loadWatchlistSnapshot() async {
+    try {
+      final map = <String, Set<String>>{};
+      var page = 1;
+      while (true) {
+        final res = await fetchWatchlistKeyPage(page, watchlistPageLimit);
+        for (final group in res.groups) {
+          final shared = group.toSet();
+          for (final key in shared) {
+            map[key] = shared;
+          }
+        }
+        if (!res.hasMore) break;
+        if (page >= watchlistMaxPages) {
+          appLogger.w('$watchlistLogLabel snapshot truncated at ${map.length} keys ($page pages)');
+          break;
+        }
+        page++;
+      }
+      _watchlistKeyGroups = map;
+      _watchlistChanges.notify();
+    } catch (e) {
+      appLogger.w('$watchlistLogLabel snapshot load failed', error: e);
+    }
+  }
+
+  bool? isOnWatchlist(MediaKind kind, CatalogItemIds ids) {
+    final map = _watchlistKeyGroups;
+    if (map == null) return null;
+    return membershipKeysFor(kind, ids).any(map.containsKey);
+  }
+
+  Future<void> addToWatchlist(MediaKind kind, CatalogItemIds ids) => _mutateWatchlist(kind, ids, add: true);
+
+  Future<void> removeFromWatchlist(MediaKind kind, CatalogItemIds ids) => _mutateWatchlist(kind, ids, add: false);
+
+  Future<void> _mutateWatchlist(MediaKind kind, CatalogItemIds ids, {required bool add}) async {
+    final resolved = await resolveWatchlistMutationIds(kind, ids);
+    final keys = membershipKeysFor(kind, resolved);
+
+    // Optimistic: flip the snapshot first so UI toggles instantly; revert on
+    // failure. Callers surface the rethrown error.
+    final map = _watchlistKeyGroups;
+    Set<String>? addedGroup;
+    List<Set<String>>? removedGroups;
+    var changed = false;
+    if (map != null && keys.isNotEmpty) {
+      if (add) {
+        addedGroup = keys.toSet();
+        for (final key in addedGroup) {
+          map[key] = addedGroup;
+        }
+        changed = true;
+      } else {
+        removedGroups = _takeGroups(map, keys);
+        changed = removedGroups.isNotEmpty;
+      }
+      if (changed) _watchlistChanges.notify();
+    }
+
+    try {
+      await performWatchlistMutation(kind, resolved, add: add);
+    } catch (_) {
+      // Revert only if the snapshot wasn't replaced by a reload meanwhile.
+      if (changed && identical(map, _watchlistKeyGroups)) {
+        if (addedGroup != null) {
+          for (final key in addedGroup) {
+            if (identical(map![key], addedGroup)) map.remove(key);
+          }
+        }
+        for (final group in removedGroups ?? const <Set<String>>[]) {
+          for (final key in group) {
+            map![key] = group;
+          }
+        }
+        _watchlistChanges.notify();
+      }
+      rethrow;
+    }
+  }
+
+  /// Remove every entry group hit by [keys] from [map], returning them for
+  /// a potential revert.
+  static List<Set<String>> _takeGroups(Map<String, Set<String>> map, Iterable<String> keys) {
+    final groups = <Set<String>>[];
+    for (final key in keys) {
+      final group = map[key];
+      if (group != null && !groups.any((existing) => identical(existing, group))) {
+        groups.add(group);
+      }
+    }
+    for (final group in groups) {
+      group.forEach(map.remove);
+    }
+    return groups;
+  }
+
+  /// Call from the source's [CatalogSource.dispose].
+  void disposeWatchlistMachinery() {
+    _watchlistChanges.dispose();
+  }
+}

lib/services/catalog/trakt_catalog_source.dart

@@ -0,0 +1,217 @@
+import '../../media/media_kind.dart';
+import '../../models/catalog/catalog_cast_member.dart';
+import '../../models/catalog/catalog_item.dart';
+import '../../models/trakt/trakt_catalog_entry.dart';
+import '../../models/trakt/trakt_catalog_media.dart';
+import '../../models/trakt/trakt_ids.dart';
+import '../../utils/external_ids.dart';
+import '../trakt/trakt_client.dart';
+import '../trakt/trakt_constants.dart';
+import 'catalog_source.dart';
+import 'catalog_watchlist_machinery.dart';
+
+/// [CatalogSource] backed by the Trakt API.
+///
+/// Wraps the catalog [TraktClient] owned by `TraktAccountProvider` (not owned
+/// here — never disposed by this class). Watchlist membership rides
+/// [CatalogWatchlistMachinery] with kind-namespaced keys over every id form.
+class TraktCatalogSource with CatalogWatchlistMachinery implements CatalogSource {
+  final TraktClient _client;
+
+  TraktCatalogSource(this._client);
+
+  @override
+  String get watchlistLogLabel => 'Trakt: watchlist';
+
+  /// Full-snapshot paging: 4 × 250 covers 1000 watchlist entries.
+  @override
+  int get watchlistPageLimit => 250;
+  @override
+  int get watchlistMaxPages => 4;
+
+  @override
+  CatalogSourceId get id => CatalogSourceId.trakt;
+
+  @override
+  String get displayName => 'Trakt';
+
+  @override
+  List<CatalogRowId> get supportedRows => const [
+    CatalogRowId.watchlist,
+    CatalogRowId.recommendedMovies,
+    CatalogRowId.recommendedShows,
+    CatalogRowId.trendingMovies,
+    CatalogRowId.trendingShows,
+    CatalogRowId.popularMovies,
+    CatalogRowId.popularShows,
+  ];
+
+  @override
+  bool get supportsWatchlist => true;
+
+  @override
+  Future<CatalogPage> fetchRow(CatalogRowId row, {int page = 1, int limit = 25}) async {
+    switch (row) {
+      case CatalogRowId.watchlist:
+        final res = await _client.getWatchlist(page: page, limit: limit);
+        return CatalogPage(items: _fromEntries(res.items), hasMore: res.hasMore);
+      case CatalogRowId.recommendedMovies:
+        return CatalogPage(
+          items: _fromMedia(await _client.getRecommended(TraktCatalogType.movies, limit: limit), MediaKind.movie),
+        );
+      case CatalogRowId.recommendedShows:
+        return CatalogPage(
+          items: _fromMedia(await _client.getRecommended(TraktCatalogType.shows, limit: limit), MediaKind.show),
+        );
+      case CatalogRowId.trendingMovies:
+        final res = await _client.getTrending(TraktCatalogType.movies, page: page, limit: limit);
+        return CatalogPage(
+          items: _fromEntries(res.items, kind: MediaKind.movie),
+          hasMore: res.hasMore,
+        );
+      case CatalogRowId.trendingShows:
+        final res = await _client.getTrending(TraktCatalogType.shows, page: page, limit: limit);
+        return CatalogPage(
+          items: _fromEntries(res.items, kind: MediaKind.show),
+          hasMore: res.hasMore,
+        );
+      case CatalogRowId.popularMovies:
+        final res = await _client.getPopular(TraktCatalogType.movies, page: page, limit: limit);
+        return CatalogPage(items: _fromMedia(res.items, MediaKind.movie), hasMore: res.hasMore);
+      case CatalogRowId.popularShows:
+        final res = await _client.getPopular(TraktCatalogType.shows, page: page, limit: limit);
+        return CatalogPage(items: _fromMedia(res.items, MediaKind.show), hasMore: res.hasMore);
+      case CatalogRowId.suggestedAnime:
+      case CatalogRowId.airingAnime:
+      case CatalogRowId.popularAnime:
+      case CatalogRowId.trending:
+      case CatalogRowId.upcomingMovies:
+      case CatalogRowId.upcomingShows:
+        throw ArgumentError('Trakt does not serve ${row.name}');
+    }
+  }
+
+  @override
+  Future<List<CatalogItem>> search(String query, {int limit = 30}) async {
+    final trimmed = query.trim();
+    if (trimmed.isEmpty) return const [];
+    final res = await _client.searchCatalog(trimmed, limit: limit);
+    return _fromEntries(res.items);
+  }
+
+  @override
+  Future<CatalogItemIds?> resolveItemIds(MediaKind kind, ExternalIds external) async =>
+      external.hasAny ? CatalogItemIds.fromExternal(external) : null;
+
+  @override
+  Future<List<CatalogCastMember>> fetchCast(CatalogItem item, {int limit = 20}) async {
+    final id = item.ids.trakt?.toString() ?? item.ids.slug ?? item.ids.imdb;
+    if (id == null) return const [];
+    final type = item.kind == MediaKind.movie ? TraktCatalogType.movies : TraktCatalogType.shows;
+    final cast = await _client.getPeople(type, id);
+    return [
+      for (final entry in cast.take(limit))
+        if (entry.person?.name case final String name when name.isNotEmpty)
+          CatalogCastMember(
+            name: name,
+            secondary: entry.characters?.firstOrNull,
+            imageUrl: entry.person?.images?.primaryHeadshot,
+          ),
+    ];
+  }
+
+  @override
+  Future<List<CatalogItem>> fetchRelated(CatalogItem item, {int limit = 20}) async {
+    final id = item.ids.trakt?.toString() ?? item.ids.slug ?? item.ids.imdb;
+    if (id == null) return const [];
+    final type = item.kind == MediaKind.movie ? TraktCatalogType.movies : TraktCatalogType.shows;
+    return _fromMedia(await _client.getRelated(type, id, limit: limit), item.kind);
+  }
+
+  @override
+  Future<WatchlistKeyPage> fetchWatchlistKeyPage(int page, int limit) async {
+    final res = await _client.getWatchlist(page: page, limit: limit);
+    return (
+      groups: [for (final item in _fromEntries(res.items)) membershipKeysFor(item.kind, item.ids)],
+      hasMore: res.hasMore,
+    );
+  }
+
+  @override
+  Future<void> performWatchlistMutation(MediaKind kind, CatalogItemIds ids, {required bool add}) async {
+    final body = {
+      kind == MediaKind.show ? 'shows' : 'movies': [
+        {'ids': TraktIds(trakt: ids.trakt, slug: ids.slug, imdb: ids.imdb, tmdb: ids.tmdb, tvdb: ids.tvdb).toJson()},
+      ],
+    };
+    add ? await _client.addToWatchlist(body) : await _client.removeFromWatchlist(body);
+  }
+
+  @override
+  List<String> membershipKeysFor(MediaKind kind, CatalogItemIds ids) => [
+    for (final key in ids.allKeys) '${kind.id}/$key',
+  ];
+
+  List<CatalogItem> _fromEntries(List<TraktCatalogEntry> entries, {MediaKind? kind}) => [
+    for (final entry in entries)
+      if (entry.media != null && _entryKind(entry, kind) != null && entry.media!.ids.hasAny)
+        _toCatalogItem(entry.media!, _entryKind(entry, kind)!),
+  ];
+
+  List<CatalogItem> _fromMedia(List<TraktCatalogMedia> media, MediaKind kind) => [
+    for (final m in media)
+      if (m.ids.hasAny) _toCatalogItem(m, kind),
+  ];
+
+  /// Watchlist entries carry a `type` field; trending entries are typed by
+  /// which wrapper key is present; fixed-kind endpoints pass [fixed].
+  static MediaKind? _entryKind(TraktCatalogEntry entry, MediaKind? fixed) {
+    if (fixed != null) return fixed;
+    return switch (entry.type) {
+      'movie' => MediaKind.movie,
+      'show' => MediaKind.show,
+      null => entry.isShow ? MediaKind.show : MediaKind.movie,
+      _ => null, // seasons/episodes on the watchlist are not Explore rows
+    };
+  }
+
+  /// Normalize Trakt's status strings. Movies' `released` maps to null —
+  /// a "Released" chip on every movie is noise.
+  static CatalogAirStatus? airStatusFor(String? status) => switch (status) {
+    'returning series' || 'continuing' => CatalogAirStatus.airing,
+    'ended' => CatalogAirStatus.ended,
+    'canceled' => CatalogAirStatus.canceled,
+    'in production' ||
+    'post production' ||
+    'planned' ||
+    'upcoming' ||
+    'pilot' ||
+    'rumored' => CatalogAirStatus.upcoming,
+    _ => null,
+  };
+
+  CatalogItem _toCatalogItem(TraktCatalogMedia m, MediaKind kind) => CatalogItem(
+    source: CatalogSourceId.trakt,
+    kind: kind,
+    title: m.title ?? '',
+    year: m.year,
+    overview: m.overview,
+    runtimeMinutes: m.runtime,
+    rating: m.rating,
+    votes: m.votes,
+    genres: m.genres,
+    certification: m.certification,
+    trailerUrl: m.trailer,
+    airStatus: airStatusFor(m.status),
+    episodeCount: m.airedEpisodes,
+    network: m.network,
+    ids: CatalogItemIds(trakt: m.ids.trakt, slug: m.ids.slug, imdb: m.ids.imdb, tmdb: m.ids.tmdb, tvdb: m.ids.tvdb),
+    posterUrl: m.images?.primaryPoster,
+    backdropUrl: m.images?.primaryBackdrop,
+  );
+
+  @override
+  void dispose() {
+    disposeWatchlistMachinery();
+  }
+}

lib/services/trakt/trakt_client.dart

@@ -3,14 +3,19 @@ import 'dart:convert';
 
 import 'package:http/http.dart' as http;
 
+import '../../models/trakt/trakt_cast_entry.dart';
+import '../../models/trakt/trakt_catalog_entry.dart';
+import '../../models/trakt/trakt_catalog_media.dart';
 import '../../models/trakt/trakt_scrobble_request.dart';
 import '../../models/trakt/trakt_user.dart';
 import '../../utils/app_logger.dart';
+import '../trackers/future_coalescer.dart';
 import '../trackers/tracker_constants.dart';
 import '../trackers/tracker_exceptions.dart';
 import '../trackers/tracker_http_client.dart';
 import '../trackers/tracker_session.dart';
 import 'trakt_constants.dart';
+import 'trakt_page.dart';
 
 /// HTTP wrapper for the Trakt REST API.
 ///
@@ -19,7 +24,7 @@ import 'trakt_constants.dart';
 class TraktClient {
   static const Set<int> _scrobbleAllowedStatuses = {200, 201, 409};
   static const Set<int> _permanentRefreshFailureStatuses = {400, 401, 403};
-  static final Map<String, Future<TrackerSession>> _refreshesByToken = {};
+  static final KeyedFutureCoalescer<String, TrackerSession> _refreshesByToken = KeyedFutureCoalescer();
 
   TrackerSession _session;
   final TrackerHttpClient _http;
@@ -78,6 +83,100 @@ class TraktClient {
     return res is List ? res : const [];
   }
 
+  // --- Catalog endpoints (Explore tab) ---
+
+  static const String _catalogExtended = 'extended=full,images';
+
+  /// `GET /sync/watchlist[/{type}/{sort}]`. A null [type] returns all entry
+  /// types mixed, in the user's rank order. Pagination is currently optional
+  /// on this endpoint; sending page/limit makes Trakt echo X-Pagination
+  /// headers.
+  Future<TraktPage<TraktCatalogEntry>> getWatchlist({
+    TraktCatalogType? type,
+    String sort = 'added',
+    int page = 1,
+    int limit = 100,
+  }) async {
+    final path = type == null ? '/sync/watchlist' : '/sync/watchlist/${type.name}/$sort';
+    final res = await _requestResponse('GET', '$path?$_catalogExtended&page=$page&limit=$limit');
+    return TraktPage.fromResponse(res, _decodeEntries(res.body));
+  }
+
+  /// Items are wrapped as `{watchers, movie|show}`. Public endpoint, but sent
+  /// authenticated: the tab only exists with a session and per-user rate
+  /// limiting is cleaner than app-level.
+  Future<TraktPage<TraktCatalogEntry>> getTrending(TraktCatalogType type, {int page = 1, int limit = 25}) async {
+    final res = await _requestResponse('GET', '/${type.name}/trending?$_catalogExtended&page=$page&limit=$limit');
+    return TraktPage.fromResponse(res, _decodeEntries(res.body));
+  }
+
+  /// Returns bare movie/show objects (not wrapped like trending).
+  Future<TraktPage<TraktCatalogMedia>> getPopular(TraktCatalogType type, {int page = 1, int limit = 25}) async {
+    final res = await _requestResponse('GET', '/${type.name}/popular?$_catalogExtended&page=$page&limit=$limit');
+    return TraktPage.fromResponse(res, _decodeMedia(res.body));
+  }
+
+  /// Personalized recommendations. OAuth-required, limit-only (no pagination).
+  Future<List<TraktCatalogMedia>> getRecommended(
+    TraktCatalogType type, {
+    int limit = 25,
+    bool ignoreCollected = false,
+    bool ignoreWatchlisted = true,
+  }) async {
+    final res = await _requestResponse(
+      'GET',
+      '/recommendations/${type.name}'
+          '?$_catalogExtended&limit=$limit&ignore_collected=$ignoreCollected&ignore_watchlisted=$ignoreWatchlisted',
+    );
+    return _decodeMedia(res.body);
+  }
+
+  /// Title search across movies and shows (`GET /search/movie,show`).
+  /// Results are wrapped `{type, score, movie|show}` like watchlist entries.
+  Future<TraktPage<TraktCatalogEntry>> searchCatalog(String query, {int page = 1, int limit = 25}) async {
+    final res = await _requestResponse(
+      'GET',
+      '/search/movie,show?query=${Uri.encodeQueryComponent(query)}&$_catalogExtended&page=$page&limit=$limit',
+    );
+    return TraktPage.fromResponse(res, _decodeEntries(res.body));
+  }
+
+  /// Similar titles (`GET /{movies|shows}/{id}/related`) — bare media
+  /// objects of the same type. [id] is a Trakt numeric id or slug.
+  Future<List<TraktCatalogMedia>> getRelated(TraktCatalogType type, String id, {int limit = 20}) async {
+    final res = await _requestResponse('GET', '/${type.name}/$id/related?$_catalogExtended&limit=$limit');
+    return _decodeMedia(res.body);
+  }
+
+  /// Cast credits of a title (`GET /{movies|shows}/{id}/people`), in billing
+  /// order. Crew is not parsed. [id] is a Trakt numeric id or slug.
+  Future<List<TraktCastEntry>> getPeople(TraktCatalogType type, String id) async {
+    final res = await _requestResponse('GET', '/${type.name}/$id/people?$_catalogExtended');
+    final decoded = TrackerHttpClient.decodeJson(res.body);
+    if (decoded is! Map) return const [];
+    final cast = decoded['cast'];
+    if (cast is! List) return const [];
+    return [for (final e in cast.whereType<Map<String, dynamic>>()) TraktCastEntry.fromJson(e)];
+  }
+
+  /// Body shape: `{"movies":[{"ids":{...}}],"shows":[{"ids":{...}}]}`.
+  Future<void> addToWatchlist(Map<String, dynamic> body) =>
+      _request('POST', '/sync/watchlist', body: body, allowStatuses: const {200, 201});
+
+  Future<void> removeFromWatchlist(Map<String, dynamic> body) => _request('POST', '/sync/watchlist/remove', body: body);
+
+  static List<TraktCatalogEntry> _decodeEntries(String body) {
+    final decoded = TrackerHttpClient.decodeJson(body);
+    if (decoded is! List) return const [];
+    return [for (final e in decoded.whereType<Map<String, dynamic>>()) TraktCatalogEntry.fromJson(e)];
+  }
+
+  static List<TraktCatalogMedia> _decodeMedia(String body) {
+    final decoded = TrackerHttpClient.decodeJson(body);
+    if (decoded is! List) return const [];
+    return [for (final e in decoded.whereType<Map<String, dynamic>>()) TraktCatalogMedia.fromJson(e)];
+  }
+
   /// Refresh the access token. Coalesces concurrent calls so
   /// duplicate POSTs don't race when multiple in-flight requests hit 401.
   Future<TrackerSession> refresh() async {
@@ -88,31 +187,26 @@ class TraktClient {
       if (e.isPermanent) onSessionInvalidated();
       rethrow;
     }
-    final existing = _refreshesByToken[refreshToken];
-    if (existing != null) {
-      try {
-        final session = await existing;
-        if (_session.refreshToken == refreshToken) {
-          _session = session;
-          onSessionUpdated?.call(session);
-        }
-        return _session;
-      } on TrackerAuthException catch (e) {
-        if (e.isPermanent && _session.refreshToken == refreshToken) {
-          onSessionInvalidated();
-        }
-        rethrow;
+    var initiated = false;
+    try {
+      final session = await _refreshesByToken.run(refreshToken, () {
+        initiated = true;
+        return _doRefresh(refreshToken);
+      });
+      // No-op for the initiating client (_doRefresh already adopted, so its
+      // refreshToken moved on); joiners sharing the token adopt here.
+      if (_session.refreshToken == refreshToken) {
+        _session = session;
+        onSessionUpdated?.call(session);
       }
+      return _session;
+    } on TrackerAuthException catch (e) {
+      // The initiator's _doRefresh already invalidated; joiners do it here.
+      if (!initiated && e.isPermanent && _session.refreshToken == refreshToken) {
+        onSessionInvalidated();
+      }
+      rethrow;
     }
-
-    late final Future<TrackerSession> refresh;
-    refresh = _doRefresh(refreshToken).whenComplete(() {
-      if (identical(_refreshesByToken[refreshToken], refresh)) {
-        _refreshesByToken.remove(refreshToken);
-      }
-    });
-    _refreshesByToken[refreshToken] = refresh;
-    return refresh;
   }
 
   Future<TrackerSession> _doRefresh(String refreshToken) async {
@@ -187,6 +281,18 @@ class TraktClient {
     String path, {
     Map<String, dynamic>? body,
     Set<int> allowStatuses = const {200, 201, 204},
+  }) async {
+    final res = await _requestResponse(method, path, body: body, allowStatuses: allowStatuses);
+    return TrackerHttpClient.decodeJson(res.body);
+  }
+
+  /// [_request] variant exposing the raw response for callers that need
+  /// headers (pagination).
+  Future<http.Response> _requestResponse(
+    String method,
+    String path, {
+    Map<String, dynamic>? body,
+    Set<int> allowStatuses = const {200, 201, 204},
   }) async {
     if (_session.needsRefresh) {
       try {
@@ -203,9 +309,7 @@ class TraktClient {
       res = await _send(method, path, body: body);
     }
 
-    if (allowStatuses.contains(res.statusCode)) {
-      return TrackerHttpClient.decodeJson(res.body);
-    }
+    if (allowStatuses.contains(res.statusCode)) return res;
 
     if (res.statusCode == 429) {
       throw TrackerRateLimitException(

lib/services/trakt/trakt_constants.dart

@@ -37,14 +37,13 @@ class TraktConstants {
   }
 }
 
-/// Build a `SharedPreferences` key scoped to the given Plex profile UUID.
-/// Mirrors the `_userPrefix` pattern in [StorageService] so each profile gets
-/// its own Trakt session and sync queue.
-String traktUserKey(String userUuid, String baseKey) => userUuid.isEmpty ? baseKey : 'user_${userUuid}_$baseKey';
-
 /// Scrobble lifecycle state sent to Trakt's `/scrobble/{name}` endpoints.
 enum TraktScrobbleState { start, pause, stop }
 
+/// Catalog list flavor for Trakt's discover/watchlist endpoints, named after
+/// the URL path segment (`/movies/trending`, `/sync/watchlist/shows/...`).
+enum TraktCatalogType { movies, shows }
+
 /// Direction of a watched-status sync push.
 enum TraktSyncOp {
   add,

lib/services/trakt/trakt_page.dart

@@ -0,0 +1,23 @@
+import 'package:http/http.dart' as http;
+
+/// One page of a paginated Trakt response, parsed from the
+/// `X-Pagination-*` headers.
+class TraktPage<T> {
+  final List<T> items;
+  final int page;
+  final int pageCount;
+  final int itemCount;
+
+  const TraktPage({required this.items, required this.page, required this.pageCount, required this.itemCount});
+
+  bool get hasMore => page < pageCount;
+
+  /// Endpoints where pagination is optional omit the headers; default to a
+  /// single page so callers never loop.
+  factory TraktPage.fromResponse(http.Response res, List<T> items) => TraktPage(
+    items: items,
+    page: int.tryParse(res.headers['x-pagination-page'] ?? '') ?? 1,
+    pageCount: int.tryParse(res.headers['x-pagination-page-count'] ?? '') ?? 1,
+    itemCount: int.tryParse(res.headers['x-pagination-item-count'] ?? '') ?? items.length,
+  );
+}

lib/services/trakt/trakt_sync_queue.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 
 import '../../models/trakt/trakt_ids.dart';
+import '../../profiles/profile.dart';
 import '../base_shared_preferences_service.dart';
 import '../../utils/app_logger.dart';
 import 'trakt_constants.dart';
@@ -97,7 +98,7 @@ class TraktSyncQueue {
 
   Future<List<TraktSyncQueueItem>> load(String userUuid) async {
     final prefs = await BaseSharedPreferencesService.sharedCache();
-    final key = traktUserKey(userUuid, _baseKey);
+    final key = profileScopedPrefsKey(userUuid, _baseKey);
     final raw = prefs.getString(key);
     if (raw == null) return [];
     try {
@@ -105,7 +106,7 @@ class TraktSyncQueue {
       return list.map((e) => TraktSyncQueueItem.fromJson(e as Map<String, dynamic>)).toList();
     } catch (e, st) {
       appLogger.e('Trakt sync queue parse failed, discarding', error: e, stackTrace: st);
-      await prefs.setString(traktUserKey(userUuid, '${_baseKey}_corrupt'), raw);
+      await prefs.setString(profileScopedPrefsKey(userUuid, '${_baseKey}_corrupt'), raw);
       await prefs.remove(key);
       return [];
     }
@@ -117,7 +118,7 @@ class TraktSyncQueue {
 
   Future<void> _saveRaw(String userUuid, List<TraktSyncQueueItem> items) async {
     final prefs = await BaseSharedPreferencesService.sharedCache();
-    final key = traktUserKey(userUuid, _baseKey);
+    final key = profileScopedPrefsKey(userUuid, _baseKey);
     if (items.isEmpty) {
       await prefs.remove(key);
     } else {