fix(trakt): publish refreshed sessions

lib/services/trakt/trakt_client.dart

@@ -10,7 +10,6 @@ import '../../utils/app_logger.dart';
 import '../../utils/platform_http_client_stub.dart'
     if (dart.library.io) '../../utils/platform_http_client_io.dart'
     as platform;
-import '../trackers/future_coalescer.dart';
 import '../trackers/tracker_constants.dart';
 import 'trakt_constants.dart';
 import 'trakt_session.dart';
@@ -21,6 +20,8 @@ import 'trakt_session.dart';
 /// coalesced so we only hit `/oauth/token` once per refresh.
 class TraktClient {
   static const Set<int> _scrobbleAllowedStatuses = {200, 201, 409};
+  static const Set<int> _permanentRefreshFailureStatuses = {400, 401, 403};
+  static final Map<String, Future<TraktSession>> _refreshesByToken = {};
 
   TraktSession _session;
   final http.Client _http;
@@ -29,14 +30,25 @@ class TraktClient {
   /// uses this to clear the stored session and notify the UI.
   final void Function() onSessionInvalidated;
 
-  final _refreshCoalescer = FutureCoalescer<TraktSession>();
+  /// Fired when refresh succeeds so the provider can persist the rotated
+  /// access/refresh token pair and share it with the other active Trakt clients.
+  final void Function(TraktSession session)? onSessionUpdated;
 
-  TraktClient(TraktSession session, {required this.onSessionInvalidated, http.Client? httpClient})
+  TraktClient(
+    TraktSession session, {
+    required this.onSessionInvalidated,
+    this.onSessionUpdated,
+    http.Client? httpClient,
+  })
     : _session = session,
       _http = httpClient ?? platform.createPlatformClient();
 
   TraktSession get session => _session;
 
+  void updateSession(TraktSession session) {
+    _session = session;
+  }
+
   void dispose() => _http.close();
 
   Future<TraktUser> getUserSettings() async {
@@ -71,9 +83,36 @@ class TraktClient {
 
   /// Refresh the access token. Coalesces concurrent calls so
   /// duplicate POSTs don't race when multiple in-flight requests hit 401.
-  Future<TraktSession> refresh() => _refreshCoalescer.run(_doRefresh);
+  Future<TraktSession> refresh() async {
+    final refreshToken = _session.refreshToken;
+    final existing = _refreshesByToken[refreshToken];
+    if (existing != null) {
+      try {
+        final session = await existing;
+        if (_session.refreshToken == refreshToken) {
+          _session = session;
+          onSessionUpdated?.call(session);
+        }
+        return _session;
+      } on TraktAuthException catch (e) {
+        if (e.isPermanent && _session.refreshToken == refreshToken) {
+          onSessionInvalidated();
+        }
+        rethrow;
+      }
+    }
 
-  Future<TraktSession> _doRefresh() async {
+    late final Future<TraktSession> refresh;
+    refresh = _doRefresh(refreshToken).whenComplete(() {
+      if (identical(_refreshesByToken[refreshToken], refresh)) {
+        _refreshesByToken.remove(refreshToken);
+      }
+    });
+    _refreshesByToken[refreshToken] = refresh;
+    return refresh;
+  }
+
+  Future<TraktSession> _doRefresh(String refreshToken) async {
     appLogger.d('Trakt: refreshing access token');
     final tokenUri = Uri.parse(TraktConstants.tokenUrl);
     final res = await sendAbortableHttpRequest(
@@ -82,7 +121,7 @@ class TraktClient {
       tokenUri,
       headers: TraktConstants.headers(),
       body: json.encode({
-        'refresh_token': _session.refreshToken,
+        'refresh_token': refreshToken,
         'client_id': TraktConstants.clientId,
         'client_secret': TraktConstants.clientSecret,
         'grant_type': 'refresh_token',
@@ -94,12 +133,27 @@ class TraktClient {
     if (res.statusCode == 200) {
       final body = json.decode(res.body) as Map<String, dynamic>;
       _session = TraktSession.fromTokenResponse(body).copyWith(username: _session.username);
+      onSessionUpdated?.call(_session);
       return _session;
     }
 
-    appLogger.w('Trakt: refresh failed (${res.statusCode}), session invalidated');
-    onSessionInvalidated();
-    throw TraktAuthException('Refresh failed: HTTP ${res.statusCode}');
+    if (_session.refreshToken != refreshToken) {
+      appLogger.d('Trakt: refresh failed (${res.statusCode}) after session update; keeping latest session');
+      return _session;
+    }
+
+    final isPermanent = _permanentRefreshFailureStatuses.contains(res.statusCode);
+    if (isPermanent) {
+      appLogger.w('Trakt: refresh failed permanently (${res.statusCode}), session invalidated');
+      onSessionInvalidated();
+    } else {
+      appLogger.w('Trakt: refresh failed (${res.statusCode}), will retry later');
+    }
+    throw TraktAuthException(
+      'Refresh failed: HTTP ${res.statusCode}',
+      statusCode: res.statusCode,
+      isPermanent: isPermanent,
+    );
   }
 
   /// Revoke the access token at Trakt. Best-effort; swallows network errors.
@@ -203,7 +257,9 @@ class TraktRateLimitException implements Exception {
 
 class TraktAuthException implements Exception {
   final String message;
-  const TraktAuthException(this.message);
+  final int? statusCode;
+  final bool isPermanent;
+  const TraktAuthException(this.message, {this.statusCode, this.isPermanent = false});
   @override
   String toString() => 'TraktAuthException: $message';
 }

lib/services/trakt/trakt_scrobble_service.dart

@@ -73,15 +73,25 @@ class TraktScrobbleService {
   void rebindToProfile(
     TraktSession? session, {
     required void Function() onSessionInvalidated,
+    void Function(TraktSession session)? onSessionUpdated,
     http.Client? httpClient,
   }) {
     _client?.dispose();
     _client = session != null
-        ? TraktClient(session, onSessionInvalidated: onSessionInvalidated, httpClient: httpClient)
+        ? TraktClient(
+            session,
+            onSessionInvalidated: onSessionInvalidated,
+            onSessionUpdated: onSessionUpdated,
+            httpClient: httpClient,
+          )
         : null;
     cancelInFlight();
   }
 
+  void updateSession(TraktSession session) {
+    _client?.updateSession(session);
+  }
+
   Future<int?> getRating(TrackerRatingContext ctx) async {
     final client = _client;
     if (client == null) throw const TrackerRatingUnavailableException('Trakt');

lib/services/trakt/trakt_sync_service.dart

@@ -80,14 +80,25 @@ class TraktSyncService {
 
   /// Switch to a different account. Drops cached resolvers (their backing
   /// clients are tied to the previous user's tokens) and rebinds the queue.
-  void rebindToProfile(String userUuid, TraktSession? session, {required void Function() onSessionInvalidated}) {
+  void rebindToProfile(
+    String userUuid,
+    TraktSession? session, {
+    required void Function() onSessionInvalidated,
+    void Function(TraktSession session)? onSessionUpdated,
+  }) {
     _client?.dispose();
-    _client = session != null ? TraktClient(session, onSessionInvalidated: onSessionInvalidated) : null;
+    _client = session != null
+        ? TraktClient(session, onSessionInvalidated: onSessionInvalidated, onSessionUpdated: onSessionUpdated)
+        : null;
     _activeUserUuid = userUuid;
     _resolvers.clear();
     if (_client != null) unawaited(flushQueue());
   }
 
+  void updateSession(TraktSession session) {
+    _client?.updateSession(session);
+  }
+
   Future<void> dispose() async {
     await _subscription?.cancel();
     _subscription = null;

test/services/trakt_client_test.dart

@@ -0,0 +1,225 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:http/http.dart' as http;
+import 'package:http/testing.dart';
+import 'package:plezy/services/trakt/trakt_client.dart';
+import 'package:plezy/services/trakt/trakt_session.dart';
+
+int _now() => DateTime.now().millisecondsSinceEpoch ~/ 1000;
+
+TraktSession _session({
+  String accessToken = 'access-old',
+  String refreshToken = 'refresh-old',
+  int? expiresAt,
+  String? username = 'alice',
+}) {
+  final now = _now();
+  return TraktSession(
+    accessToken: accessToken,
+    refreshToken: refreshToken,
+    expiresAt: expiresAt ?? now - 60,
+    scope: 'public',
+    createdAt: now - 3600,
+    username: username,
+  );
+}
+
+String _tokenBody({String accessToken = 'access-new', String refreshToken = 'refresh-new'}) {
+  return json.encode({
+    'access_token': accessToken,
+    'refresh_token': refreshToken,
+    'expires_in': 86400,
+    'scope': 'public',
+    'created_at': _now(),
+  });
+}
+
+void main() {
+  group('TraktClient refresh', () {
+    test('publishes refreshed tokens before retrying the API request', () async {
+      final updates = <TraktSession>[];
+      final requests = <http.Request>[];
+      final client = TraktClient(
+        _session(),
+        onSessionInvalidated: () => fail('refresh should not invalidate the session'),
+        onSessionUpdated: updates.add,
+        httpClient: MockClient((request) async {
+          requests.add(request);
+          if (request.url.path == '/oauth/token') {
+            expect(json.decode(request.body), containsPair('refresh_token', 'refresh-old'));
+            return http.Response(_tokenBody(), 200);
+          }
+          if (request.url.path == '/users/settings') {
+            expect(request.headers['Authorization'], 'Bearer access-new');
+            return http.Response(
+              json.encode({
+                'user': {'username': 'alice'},
+              }),
+              200,
+            );
+          }
+          fail('Unexpected request: ${request.method} ${request.url}');
+        }),
+      );
+
+      await client.getUserSettings();
+
+      expect(requests.map((r) => r.url.path), ['/oauth/token', '/users/settings']);
+      expect(updates, hasLength(1));
+      expect(updates.single.accessToken, 'access-new');
+      expect(updates.single.refreshToken, 'refresh-new');
+      expect(updates.single.username, 'alice');
+      expect(client.session.accessToken, 'access-new');
+
+      client.dispose();
+    });
+
+    test('uses a newer broadcast session when another client refreshed first', () async {
+      final releaseRefreshResponse = Completer<void>();
+      var invalidated = 0;
+      late final TraktClient client;
+      client = TraktClient(
+        _session(),
+        onSessionInvalidated: () => invalidated++,
+        onSessionUpdated: (_) {},
+        httpClient: MockClient((request) async {
+          expect(request.url.path, '/oauth/token');
+          expect(json.decode(request.body), containsPair('refresh_token', 'refresh-old'));
+          await releaseRefreshResponse.future;
+          return http.Response(json.encode({'error': 'invalid_grant'}), 400);
+        }),
+      );
+
+      final refresh = client.refresh();
+      client.updateSession(_session(accessToken: 'access-new', refreshToken: 'refresh-new', expiresAt: _now() + 86400));
+      releaseRefreshResponse.complete();
+
+      final session = await refresh;
+
+      expect(session.accessToken, 'access-new');
+      expect(session.refreshToken, 'refresh-new');
+      expect(invalidated, 0);
+
+      client.dispose();
+    });
+
+    test('coalesces simultaneous refreshes for the same refresh token across clients', () async {
+      final releaseRefreshResponse = Completer<void>();
+      final updates = <String>[];
+      var refreshPosts = 0;
+      Future<http.Response> handleRequest(http.Request request) async {
+        expect(request.url.path, '/oauth/token');
+        refreshPosts++;
+        await releaseRefreshResponse.future;
+        return http.Response(_tokenBody(), 200);
+      }
+
+      final first = TraktClient(
+        _session(),
+        onSessionInvalidated: () => fail('first client should not invalidate'),
+        onSessionUpdated: (session) => updates.add('first:${session.refreshToken}'),
+        httpClient: MockClient(handleRequest),
+      );
+      final second = TraktClient(
+        _session(),
+        onSessionInvalidated: () => fail('second client should not invalidate'),
+        onSessionUpdated: (session) => updates.add('second:${session.refreshToken}'),
+        httpClient: MockClient(handleRequest),
+      );
+
+      final firstRefresh = first.refresh();
+      final secondRefresh = second.refresh();
+      await Future<void>.delayed(Duration.zero);
+      releaseRefreshResponse.complete();
+
+      final sessions = await Future.wait([firstRefresh, secondRefresh]);
+
+      expect(refreshPosts, 1);
+      expect(sessions.map((s) => s.refreshToken), ['refresh-new', 'refresh-new']);
+      expect(updates, ['first:refresh-new', 'second:refresh-new']);
+
+      first.dispose();
+      second.dispose();
+    });
+
+    test('keeps the session connected after retryable refresh failures', () async {
+      var invalidated = 0;
+      final client = TraktClient(
+        _session(),
+        onSessionInvalidated: () => invalidated++,
+        onSessionUpdated: (_) {},
+        httpClient: MockClient((request) async => http.Response('temporary outage', 500)),
+      );
+
+      await expectLater(client.refresh(), throwsA(isA<TraktAuthException>()));
+
+      expect(invalidated, 0);
+      expect(client.session.refreshToken, 'refresh-old');
+
+      client.dispose();
+    });
+
+    test('invalidates the session after permanent refresh failures', () async {
+      var invalidated = 0;
+      final client = TraktClient(
+        _session(),
+        onSessionInvalidated: () => invalidated++,
+        onSessionUpdated: (_) => fail('failed refresh should not publish a session'),
+        httpClient: MockClient((request) async => http.Response(json.encode({'error': 'invalid_grant'}), 400)),
+      );
+
+      await expectLater(client.refresh(), throwsA(isA<TraktAuthException>()));
+
+      expect(invalidated, 1);
+      expect(client.session.refreshToken, 'refresh-old');
+
+      client.dispose();
+    });
+
+    test('coalesced waiters invalidate after permanent refresh failures', () async {
+      final releaseRefreshResponse = Completer<void>();
+      var ownerInvalidated = 0;
+      var waiterInvalidated = 0;
+      var refreshPosts = 0;
+
+      Future<http.Response> handleRequest(http.Request request) async {
+        expect(request.url.path, '/oauth/token');
+        refreshPosts++;
+        await releaseRefreshResponse.future;
+        return http.Response(json.encode({'error': 'invalid_grant'}), 400);
+      }
+
+      final owner = TraktClient(
+        _session(),
+        onSessionInvalidated: () => ownerInvalidated++,
+        onSessionUpdated: (_) => fail('failed refresh should not publish owner session'),
+        httpClient: MockClient(handleRequest),
+      );
+      final waiter = TraktClient(
+        _session(),
+        onSessionInvalidated: () => waiterInvalidated++,
+        onSessionUpdated: (_) => fail('failed refresh should not publish waiter session'),
+        httpClient: MockClient(handleRequest),
+      );
+
+      final ownerRefresh = owner.refresh();
+      final waiterRefresh = waiter.refresh();
+      final ownerExpectation = expectLater(ownerRefresh, throwsA(isA<TraktAuthException>()));
+      final waiterExpectation = expectLater(waiterRefresh, throwsA(isA<TraktAuthException>()));
+      await Future<void>.delayed(Duration.zero);
+      releaseRefreshResponse.complete();
+
+      await ownerExpectation;
+      await waiterExpectation;
+
+      expect(refreshPosts, 1);
+      expect(ownerInvalidated, 1);
+      expect(waiterInvalidated, 1);
+
+      owner.dispose();
+      waiter.dispose();
+    });
+  });
+}