edde746
094be1fa3e
Marking a movie or episode watched left it sitting in Continue Watching with a checkmark, and the only way to shift it was to play it and skip to the end. Continue Watching membership on a MediaBrowser server is derived from UserData.PlaybackPositionTicks alone; Played is never consulted. Marking played normally zeroes that position as a side effect, so the row usually disappears and nothing ever checked that it had. When something writes a position back afterwards the item is left played *and* resumable, which the resume route happily keeps returning forever. markWatched now reads the UserItemDataDto the mark already returns and clears the bookmark itself when the server left one behind, so the postcondition holds however the item got into that state. The follow-up write costs a request only when the invariant is actually broken. The writer putting items there is our own offline queue. insertWatchAction already drops queued progress for an item when the mark is itself queued, but the online mark writes straight to the server and queues nothing, so a progress row recorded earlier survived and replayed afterwards — pending actions go out oldest first — restoring the very position the mark had cleared. The sync service now listens for watch-state events and discards queued progress for that item as the mark lands. Progress recorded after a mark is a rewatch and is queued later, so it is untouched. Plex never showed this because it forwards the recorded-at timestamp and lets the server discard a stale replay; the MediaBrowser stop report has nowhere to put one. Continue Watching also drops the row locally now instead of waiting a round trip for the refetch to confirm it, matching what removal events already did, and marking a season or show takes its on-deck episode with it. Watched items are deliberately still not filtered out of the shelf: Jellyfin keeps Played set when new progress arrives, so a rewatch in progress is indistinguishable from a stuck row, and filtering would hide it. close #1812
1573 lines
57 KiB
Dart
1573 lines
57 KiB
Dart
import 'package:drift/native.dart';
|
|
import 'package:plezy/media/ids.dart';
|
|
import 'package:flutter/foundation.dart';
|
|
import 'package:flutter_test/flutter_test.dart';
|
|
import 'package:http/http.dart' as http;
|
|
import 'package:http/testing.dart';
|
|
import 'package:plezy/connection/connection.dart';
|
|
import 'package:plezy/database/app_database.dart';
|
|
import 'package:plezy/database/download_operations.dart';
|
|
import 'package:plezy/media/media_backend.dart';
|
|
import 'package:plezy/media/media_item.dart';
|
|
import 'package:plezy/media/media_kind.dart';
|
|
import 'package:plezy/media/media_server_client.dart';
|
|
import 'package:plezy/media/playback_report_metadata.dart';
|
|
import 'package:plezy/services/jellyfin_api_cache.dart';
|
|
import 'package:plezy/services/jellyfin_client.dart';
|
|
import 'package:plezy/services/multi_server_manager.dart';
|
|
import 'package:plezy/services/offline_mode_source.dart';
|
|
import 'package:plezy/services/offline_watch_sync_service.dart';
|
|
import 'package:plezy/services/plex_client.dart';
|
|
import 'package:plezy/utils/active_client_scope.dart';
|
|
import 'package:plezy/utils/watch_state_notifier.dart';
|
|
|
|
import '../test_helpers/backend_client_fixtures.dart';
|
|
import '../test_helpers/download_fixtures.dart';
|
|
import '../test_helpers/playback_report_fakes.dart';
|
|
import '../test_helpers/prefs.dart';
|
|
import '../test_helpers/media_items.dart';
|
|
|
|
// Direct `syncPendingItems` coverage exercises retry retention, Plex/Jellyfin
|
|
// progress replay, profile interruption, and scoped Jellyfin routing. Direct
|
|
// `syncWatchStatesFromServer` coverage exercises active-profile and
|
|
// active-scope routing plus selected watched outcomes. Trigger coalescing and
|
|
// throttle sequencing inside `_performBidirectionalSync`, direct cache-row and
|
|
// refresh-callback assertions, and non-default watched-threshold sources remain
|
|
// outside this suite.
|
|
|
|
/// Minimal [OfflineModeSource] that lets tests flip the offline flag and
|
|
/// observe `addListener`/`removeListener` traffic via the protected
|
|
/// [ChangeNotifier.hasListeners] flag.
|
|
class _FakeOfflineModeSource extends ChangeNotifier implements OfflineModeSource {
|
|
bool _isOffline;
|
|
_FakeOfflineModeSource({bool initial = false}) : _isOffline = initial;
|
|
|
|
@override
|
|
bool get isOffline => _isOffline;
|
|
|
|
void setOffline(bool value) {
|
|
if (_isOffline == value) return;
|
|
_isOffline = value;
|
|
notifyListeners();
|
|
}
|
|
|
|
// ChangeNotifier.hasListeners is `@protected` — re-export for tests.
|
|
@override
|
|
// ignore: unnecessary_overrides
|
|
bool get hasListeners => super.hasListeners;
|
|
}
|
|
|
|
class _RecordingMediaClient with PlaybackReportRecorder implements MediaServerClient {
|
|
_RecordingMediaClient({required this.serverId, required this.backend});
|
|
|
|
@override
|
|
final ServerId serverId;
|
|
|
|
@override
|
|
final MediaBackend backend;
|
|
|
|
@override
|
|
double get watchedThreshold => 0.9;
|
|
|
|
// Mirror the real clients: Jellyfin marks played from the stopped report, so
|
|
// the auto-scrobble path emits only the local watch event (#1287); Plex needs
|
|
// the explicit markWatched.
|
|
@override
|
|
bool get marksWatchedOnPlaybackStopped => backend == MediaBackend.jellyfin;
|
|
|
|
@override
|
|
void close() {}
|
|
|
|
final started = <({String itemId, int positionMs, int? durationMs})>[];
|
|
final stopped = <({String itemId, int positionMs, int? durationMs, PlaybackReportMetadata report})>[];
|
|
final watched = <String>[];
|
|
|
|
@override
|
|
Future<MediaItem?> fetchItem(String id) async =>
|
|
testMediaItem(id: id, backend: backend, kind: MediaKind.movie, serverId: serverId);
|
|
|
|
@override
|
|
Future<void> onPlaybackReport(PlaybackReportCall call) async {
|
|
switch (call.kind) {
|
|
case PlaybackReportKind.started:
|
|
started.add((
|
|
itemId: call.itemId,
|
|
positionMs: call.position.inMilliseconds,
|
|
durationMs: call.duration?.inMilliseconds,
|
|
));
|
|
case PlaybackReportKind.progress:
|
|
throw UnimplementedError();
|
|
case PlaybackReportKind.stopped:
|
|
stopped.add((
|
|
itemId: call.itemId,
|
|
positionMs: call.position.inMilliseconds,
|
|
durationMs: call.duration?.inMilliseconds,
|
|
report: call.report,
|
|
));
|
|
}
|
|
}
|
|
|
|
@override
|
|
Future<void> markWatched(MediaItem item) async {
|
|
watched.add(item.id);
|
|
WatchStateNotifier().notifyWatched(item: item, isNowWatched: true);
|
|
}
|
|
|
|
@override
|
|
dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
|
|
}
|
|
|
|
class _ScopedRecordingMediaClient extends _RecordingMediaClient implements ScopedMediaServerClient {
|
|
_ScopedRecordingMediaClient({required super.serverId, required super.backend, required this.scopedServerId});
|
|
|
|
@override
|
|
final String scopedServerId;
|
|
}
|
|
|
|
class _RecordingPlexClient extends _RecordingMediaClient implements PlexClient, ScopedMediaServerClient {
|
|
_RecordingPlexClient({required super.serverId, required String profileId})
|
|
: profileScopeId = buildPlexProfileScopeId(serverId: serverId, profileId: profileId),
|
|
super(backend: MediaBackend.plex);
|
|
|
|
@override
|
|
PlexProfileScopeId profileScopeId;
|
|
|
|
@override
|
|
String get scopedServerId => profileScopeId;
|
|
|
|
@override
|
|
Future<void> closeGracefully({Duration drainTimeout = const Duration(seconds: 2)}) async {}
|
|
}
|
|
|
|
/// Build a service against an in-memory database and a bare-metal
|
|
/// [MultiServerManager] (no servers added).
|
|
({OfflineWatchSyncService svc, AppDatabase db, MultiServerManager mgr}) _makeService() {
|
|
final db = AppDatabase.forTesting(NativeDatabase.memory());
|
|
JellyfinApiCache.initialize(db);
|
|
final mgr = MultiServerManager();
|
|
final svc = OfflineWatchSyncService(database: db, serverManager: mgr);
|
|
return (svc: svc, db: db, mgr: mgr);
|
|
}
|
|
|
|
JellyfinConnection _jellyfinConnection(String userId) => testJellyfinConnection(
|
|
machineId: 'jf-machine',
|
|
userId: userId,
|
|
serverName: 'Shared JF',
|
|
userName: userId,
|
|
accessToken: 'token-$userId',
|
|
deviceId: 'device',
|
|
createdAt: DateTime.fromMillisecondsSinceEpoch(0),
|
|
);
|
|
|
|
void main() {
|
|
setUp(resetSharedPreferencesForTest);
|
|
|
|
// ============================================================
|
|
// Initial state
|
|
// ============================================================
|
|
|
|
group('initial state', () {
|
|
test('a freshly constructed service is not syncing and has no pending count', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
expect(svc.isSyncing, isFalse);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
expect(await svc.getLocalWatchStatus('srv:nonexistent'), isNull);
|
|
expect(await svc.getLocalViewOffset('srv:nonexistent'), isNull);
|
|
});
|
|
|
|
test('isWatchedByProgress: pure math (no DB / network)', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
// duration=0 short-circuits to false (avoids divide-by-zero).
|
|
expect(svc.isWatchedByProgress(0, 0), isFalse);
|
|
expect(svc.isWatchedByProgress(1000, 0), isFalse);
|
|
|
|
// No serverId → uses 0.9 default threshold.
|
|
expect(svc.isWatchedByProgress(89, 100), isFalse);
|
|
expect(svc.isWatchedByProgress(90, 100), isTrue);
|
|
expect(svc.isWatchedByProgress(95, 100), isTrue);
|
|
expect(svc.isWatchedByProgress(100, 100), isTrue);
|
|
});
|
|
|
|
test('getWatchedThreshold falls back to default 0.9 when no client + no settings', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
// No SettingsService initialized, no client registered → default 90/100.
|
|
expect(svc.getWatchedThreshold(ServerId('unknown-server')), 0.9);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// queueMarkWatched / queueMarkUnwatched
|
|
// ============================================================
|
|
|
|
group('queueMarkWatched / queueMarkUnwatched', () {
|
|
test('queueMarkWatched persists a "watched" action and bumps pending count', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
var notifications = 0;
|
|
svc.addListener(() => notifications++);
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
|
|
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
// ChangeNotifier emission was synchronous in the queue helper.
|
|
expect(notifications, 1);
|
|
|
|
// Latest action has actionType='watched'.
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action, isNotNull);
|
|
expect(action!.actionType, 'watched');
|
|
expect(action.serverId, 'srv');
|
|
expect(action.ratingKey, '42');
|
|
});
|
|
|
|
test('queueMarkUnwatched persists an "unwatched" action', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '42');
|
|
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action, isNotNull);
|
|
expect(action!.actionType, 'unwatched');
|
|
});
|
|
|
|
test('queueing the opposite action replaces the prior action (single row)', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
|
|
// The DB layer's insertWatchAction deletes any prior entries for the
|
|
// same globalKey before inserting — so flipping watched/unwatched keeps
|
|
// a single row.
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '42');
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action!.actionType, 'unwatched');
|
|
});
|
|
|
|
test('concurrent watched then unwatched leaves exactly one unwatched action', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('profile-a');
|
|
|
|
final watched = svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
|
|
final unwatched = svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '42');
|
|
await Future.wait([watched, unwatched]);
|
|
|
|
final rows = (await db.getPendingWatchActions())
|
|
.where((row) => row.profileId == 'profile-a' && row.globalKey == 'srv:42')
|
|
.toList();
|
|
expect(rows, hasLength(1));
|
|
expect(rows.single.actionType, 'unwatched');
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
expect(await svc.getLocalWatchStatus('srv:42'), isFalse);
|
|
});
|
|
|
|
test('different ratingKeys persist independently', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '2');
|
|
await svc.queueMarkUnwatched(serverId: ServerId('other'), itemId: '1');
|
|
|
|
expect(await svc.getPendingSyncCount(), 3);
|
|
|
|
expect((await db.getLatestWatchAction('srv:1'))!.actionType, 'watched');
|
|
expect((await db.getLatestWatchAction('srv:2'))!.actionType, 'watched');
|
|
expect((await db.getLatestWatchAction('other:1'))!.actionType, 'unwatched');
|
|
});
|
|
});
|
|
|
|
group('syncPendingItems retry preservation', () {
|
|
test('server unavailable keeps queued action without consuming attempts', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action, isNotNull);
|
|
expect(action!.syncAttempts, 0);
|
|
expect(action.lastError, isNull);
|
|
});
|
|
|
|
test('max-attempt action is retained for explicit cleanup instead of deleted', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
|
|
var action = await db.getLatestWatchAction('srv:42');
|
|
for (var i = 0; i < OfflineWatchSyncService.maxSyncAttempts; i++) {
|
|
await db.updateSyncAttempt(action!.id, 'server error');
|
|
action = await db.getLatestWatchAction('srv:42');
|
|
}
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
final retained = await db.getLatestWatchAction('srv:42');
|
|
expect(retained, isNotNull);
|
|
expect(retained!.syncAttempts, OfflineWatchSyncService.maxSyncAttempts);
|
|
expect(retained.lastError, 'server error');
|
|
});
|
|
|
|
test('partial Plex offline progress replays as offline stopped progress', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('p1');
|
|
|
|
final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
|
|
mgr.debugRegisterClientForTesting(client);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50000, duration: 100000);
|
|
final queued = await db.getLatestWatchAction('srv:42');
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(client.started, hasLength(1));
|
|
expect(client.started.single.positionMs, 50000);
|
|
expect(client.stopped, hasLength(1));
|
|
expect(client.stopped.single.positionMs, 50000);
|
|
expect(client.stopped.single.report.isOfflineReplay, isTrue);
|
|
expect(client.stopped.single.report.recordedAt?.millisecondsSinceEpoch, queued!.updatedAt);
|
|
expect(client.watched, isEmpty);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
});
|
|
|
|
test('unknown-duration offline progress still replays stopped position', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('p1');
|
|
|
|
final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
|
|
mgr.debugRegisterClientForTesting(client);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50000, duration: null);
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(client.started, hasLength(1));
|
|
expect(client.started.single.positionMs, 50000);
|
|
expect(client.started.single.durationMs, isNull);
|
|
expect(client.stopped, hasLength(1));
|
|
expect(client.stopped.single.positionMs, 50000);
|
|
expect(client.stopped.single.durationMs, isNull);
|
|
expect(client.watched, isEmpty);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
});
|
|
|
|
test('completed Plex offline progress replays at duration and marks watched', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('p1');
|
|
|
|
final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
|
|
mgr.debugRegisterClientForTesting(client);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 95000, duration: 100000);
|
|
final queued = await db.getLatestWatchAction('srv:42');
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(client.started, isEmpty);
|
|
expect(client.stopped, hasLength(1));
|
|
expect(client.stopped.single.positionMs, 100000);
|
|
expect(client.stopped.single.durationMs, 100000);
|
|
expect(client.stopped.single.report.isOfflineReplay, isTrue);
|
|
expect(client.stopped.single.report.willContinue, isFalse);
|
|
expect(client.stopped.single.report.recordedAt?.millisecondsSinceEpoch, queued!.updatedAt);
|
|
expect(client.watched, ['42']);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
});
|
|
|
|
test('completed Jellyfin offline progress marks watched via the stop report, not markWatched (#1287)', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('p1');
|
|
|
|
final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.jellyfin);
|
|
mgr.debugRegisterClientForTesting(client);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 95000, duration: 100000);
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
// The stopped report at full duration marks the item played server-side…
|
|
expect(client.stopped, hasLength(1));
|
|
expect(client.stopped.single.positionMs, 100000);
|
|
// …so the offline replay must NOT also call markWatched — that would
|
|
// double-scrobble through the Jellyfin Trakt plugin.
|
|
expect(client.watched, isEmpty);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// queueProgressUpdate (also exercised so we can test the progress branches
|
|
// of getLocalWatchStatus / getLocalViewOffset).
|
|
// ============================================================
|
|
|
|
group('syncPendingItems profile scoping', () {
|
|
test('defers entirely when no profile is active', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
svc.setActiveProfileId('p1');
|
|
final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex);
|
|
mgr.debugRegisterClientForTesting(client);
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42');
|
|
|
|
// Active profile cleared (sign-out/teardown window): replaying the
|
|
// queue through whatever clients are bound would hit the wrong user.
|
|
svc.setActiveProfileId(null);
|
|
await svc.syncPendingItems();
|
|
|
|
expect(client.watched, isEmpty);
|
|
expect(await db.getPendingSyncCount(), 1);
|
|
});
|
|
|
|
test('requeues remaining actions when the active profile changes mid-sync', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
svc.setActiveProfileId('p1');
|
|
final posts = <String>[];
|
|
final client = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-a'),
|
|
httpClient: MockClient((request) async {
|
|
if (request.method == 'GET' && request.url.path.startsWith('/Users/user-a/Items/')) {
|
|
final id = request.url.pathSegments.last;
|
|
return http.Response('{"Id":"$id","Type":"Movie","Name":"Movie $id"}', 200);
|
|
}
|
|
if (request.method == 'POST' && request.url.path.startsWith('/UserPlayedItems/')) {
|
|
posts.add(request.url.path);
|
|
// The switch lands while action 1 is mid-flight — the binder
|
|
// would now be rebinding this server id to another user.
|
|
svc.setActiveProfileId('p2');
|
|
return http.Response('', 204);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
addTearDown(client.close);
|
|
mgr.debugRegisterJellyfinClientForTesting(client);
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1');
|
|
await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-2');
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(posts, hasLength(1));
|
|
expect(await db.getPendingSyncCount(), 1);
|
|
});
|
|
});
|
|
|
|
group('queueProgressUpdate', () {
|
|
test('persists a progress row with shouldMarkWatched=false below threshold', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
// 50% progress → below default 0.9 threshold.
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50, duration: 100);
|
|
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action, isNotNull);
|
|
expect(action!.actionType, 'progress');
|
|
expect(action.viewOffset, 50);
|
|
expect(action.duration, 100);
|
|
expect(action.shouldMarkWatched, isFalse);
|
|
});
|
|
|
|
test('persists unknown-duration progress without marking watched', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50, duration: null);
|
|
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action, isNotNull);
|
|
expect(action!.actionType, 'progress');
|
|
expect(action.viewOffset, 50);
|
|
expect(action.duration, isNull);
|
|
expect(action.shouldMarkWatched, isFalse);
|
|
});
|
|
|
|
test('persists shouldMarkWatched=true at/above the default 0.9 threshold', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 95, duration: 100);
|
|
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action!.shouldMarkWatched, isTrue);
|
|
});
|
|
|
|
test('repeated progress updates merge into the same row', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 10, duration: 100);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 20, duration: 100);
|
|
|
|
// upsertProgressAction merges by globalKey — only ONE row.
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
final action = await db.getLatestWatchAction('srv:42');
|
|
expect(action!.viewOffset, 20);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// Superseded queued progress (#1812)
|
|
// ============================================================
|
|
|
|
group('queued progress superseded by a watch-state write', () {
|
|
MediaItem itemFor(String id) =>
|
|
testMediaItem(id: id, backend: MediaBackend.jellyfin, kind: MediaKind.movie, serverId: 'srv');
|
|
|
|
// The online mark writes straight to the server and queues nothing, so
|
|
// nothing purges the queue the way insertWatchAction does for the offline
|
|
// mark. Replaying the stale row afterwards rewrites the resume position the
|
|
// mark cleared, which pins the item to Continue Watching on MediaBrowser.
|
|
test('a watched event drops the queued progress row for that item', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50000, duration: 100000);
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
|
|
WatchStateNotifier().notifyWatched(item: itemFor('42'));
|
|
await pumpEventQueue();
|
|
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
expect(await db.getLatestWatchAction('srv:42'), isNull);
|
|
});
|
|
|
|
test('an unwatched event drops it too', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50000, duration: 100000);
|
|
|
|
WatchStateNotifier().notifyWatched(item: itemFor('42'), isNowWatched: false);
|
|
await pumpEventQueue();
|
|
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
});
|
|
|
|
test('other items and queued manual marks are left alone', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50000, duration: 100000);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '43', viewOffset: 50000, duration: 100000);
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '44');
|
|
|
|
WatchStateNotifier().notifyWatched(item: itemFor('42'));
|
|
await pumpEventQueue();
|
|
|
|
expect(await db.getLatestWatchAction('srv:42'), isNull);
|
|
expect((await db.getLatestWatchAction('srv:43'))?.actionType, 'progress');
|
|
expect((await db.getLatestWatchAction('srv:44'))?.actionType, 'watched');
|
|
});
|
|
|
|
test('progress recorded after the mark survives — that is a rewatch', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
WatchStateNotifier().notifyWatched(item: itemFor('42'));
|
|
await pumpEventQueue();
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 5000, duration: 100000);
|
|
|
|
expect((await db.getLatestWatchAction('srv:42'))?.viewOffset, 5000);
|
|
});
|
|
|
|
test('a superseded row never reaches the server on the next sync', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('p1');
|
|
|
|
final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.jellyfin);
|
|
mgr.debugRegisterClientForTesting(client);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '42', viewOffset: 50000, duration: 100000);
|
|
|
|
WatchStateNotifier().notifyWatched(item: itemFor('42'));
|
|
await pumpEventQueue();
|
|
await svc.syncPendingItems();
|
|
|
|
// The whole point: no position write, so the resume bookmark the mark
|
|
// cleared stays cleared.
|
|
expect(client.stopped, isEmpty);
|
|
expect(client.started, isEmpty);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// getLocalWatchStatus
|
|
// ============================================================
|
|
|
|
group('getLocalWatchStatus', () {
|
|
test('returns null when no local action exists', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
expect(await svc.getLocalWatchStatus('srv:none'), isNull);
|
|
});
|
|
|
|
test('returns true for a "watched" action', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
expect(await svc.getLocalWatchStatus('srv:1'), isTrue);
|
|
});
|
|
|
|
test('returns false for an "unwatched" action', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '1');
|
|
expect(await svc.getLocalWatchStatus('srv:1'), isFalse);
|
|
});
|
|
|
|
test('returns watched status only for explicit actions or threshold-crossing progress', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
// Below threshold is resume-only; it must not override stale watched metadata.
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '1', viewOffset: 50, duration: 100);
|
|
expect(await svc.getLocalWatchStatus('srv:1'), isNull);
|
|
|
|
// Above threshold → shouldMarkWatched=true → status=true.
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '2', viewOffset: 99, duration: 100);
|
|
expect(await svc.getLocalWatchStatus('srv:2'), isTrue);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// getLocalViewOffset
|
|
// ============================================================
|
|
|
|
group('getLocalViewOffset', () {
|
|
test('returns null when no local action exists', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
expect(await svc.getLocalViewOffset('srv:none'), isNull);
|
|
});
|
|
|
|
test('returns null for a "watched" or "unwatched" action (no offset)', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
expect(await svc.getLocalViewOffset('srv:1'), isNull);
|
|
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '2');
|
|
expect(await svc.getLocalViewOffset('srv:2'), isNull);
|
|
});
|
|
|
|
test('returns the stored offset for a "progress" action', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '1', viewOffset: 12345, duration: 60000);
|
|
expect(await svc.getLocalViewOffset('srv:1'), 12345);
|
|
});
|
|
|
|
test('progress is replaced by a manual "watched" action — offset becomes null', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '1', viewOffset: 5000, duration: 10000);
|
|
expect(await svc.getLocalViewOffset('srv:1'), 5000);
|
|
|
|
// Manual "watched" wipes the progress row (insertWatchAction deletes
|
|
// by globalKey first), so getLocalViewOffset reads the new row whose
|
|
// actionType != 'progress' → null.
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
expect(await svc.getLocalViewOffset('srv:1'), isNull);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// getPendingSyncCount
|
|
// ============================================================
|
|
|
|
group('getPendingSyncCount', () {
|
|
test('counts every queued action (manual + progress)', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '2');
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '3', viewOffset: 50, duration: 100);
|
|
expect(await svc.getPendingSyncCount(), 3);
|
|
});
|
|
|
|
test('progress upsert does NOT increment count', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '1', viewOffset: 10, duration: 100);
|
|
await svc.queueProgressUpdate(serverId: ServerId('srv'), itemId: '1', viewOffset: 20, duration: 100);
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// getLocalWatchStatusesBatched
|
|
// ============================================================
|
|
|
|
group('getLocalWatchStatusesBatched', () {
|
|
test('empty input returns empty map without touching the DB', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
expect(await svc.getLocalWatchStatusesBatched({}), isEmpty);
|
|
});
|
|
|
|
test('returns null for missing keys, statuses for queued items', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '2');
|
|
await svc.queueProgressUpdate(
|
|
serverId: ServerId('srv'),
|
|
itemId: '3',
|
|
viewOffset: 99,
|
|
duration: 100, // above threshold
|
|
);
|
|
|
|
final result = await svc.getLocalWatchStatusesBatched({'srv:1', 'srv:2', 'srv:3', 'srv:missing'});
|
|
expect(result['srv:1'], isTrue);
|
|
expect(result['srv:2'], isFalse);
|
|
expect(result['srv:3'], isTrue);
|
|
expect(result['srv:missing'], isNull);
|
|
// The map MUST contain every requested key, even when null.
|
|
expect(result.keys.toSet(), {'srv:1', 'srv:2', 'srv:3', 'srv:missing'});
|
|
});
|
|
|
|
test('filters batched local statuses by active Jellyfin scope', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final activeUserB = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-b'),
|
|
httpClient: MockClient((_) async => http.Response('{}', 200)),
|
|
);
|
|
addTearDown(activeUserB.close);
|
|
mgr.debugRegisterJellyfinClientForTesting(activeUserB);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
await db.insertWatchAction(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
actionType: OfflineActionType.unwatched.id,
|
|
);
|
|
await Future<void>.delayed(const Duration(milliseconds: 2));
|
|
await db.insertWatchAction(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-b',
|
|
ratingKey: 'item-1',
|
|
actionType: OfflineActionType.watched.id,
|
|
);
|
|
|
|
final result = await svc.getLocalWatchStatusesBatched({'jf-machine:item-1'});
|
|
expect(result['jf-machine:item-1'], isTrue);
|
|
});
|
|
|
|
test('local watch actions are isolated by active profile', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
svc.setActiveProfileId('profile-a');
|
|
await svc.queueMarkWatched(serverId: ServerId('plex-machine'), itemId: 'item-1');
|
|
expect(await svc.getLocalWatchStatus('plex-machine:item-1'), isTrue);
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
|
|
svc.setActiveProfileId('profile-b');
|
|
expect(await svc.getLocalWatchStatus('plex-machine:item-1'), isNull);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
await svc.queueMarkUnwatched(serverId: ServerId('plex-machine'), itemId: 'item-1');
|
|
expect(await svc.getLocalWatchStatus('plex-machine:item-1'), isFalse);
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
|
|
svc.setActiveProfileId('profile-a');
|
|
expect(await svc.getLocalWatchStatus('plex-machine:item-1'), isTrue);
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
});
|
|
});
|
|
|
|
group('Plex scoped sync', () {
|
|
test('queues and replays through the exact active Plex profile scope', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('profile-a');
|
|
final clientA = _RecordingPlexClient(serverId: ServerId('plex-machine'), profileId: 'profile-a');
|
|
mgr.debugRegisterClientForTesting(clientA);
|
|
|
|
final queuedScope = await svc.queueMarkWatched(serverId: ServerId('plex-machine'), itemId: 'item-1');
|
|
expect(queuedScope, clientA.profileScopeId);
|
|
expect((await db.getPendingWatchActions()).single.clientScopeId, clientA.profileScopeId);
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(clientA.watched, ['item-1']);
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
});
|
|
|
|
test('does not replay a queued Plex owner action through a foreign active profile', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('profile-a');
|
|
final scopeA = buildPlexProfileScopeId(serverId: ServerId('plex-machine'), profileId: 'profile-a');
|
|
final clientB = _RecordingPlexClient(serverId: ServerId('plex-machine'), profileId: 'profile-b');
|
|
mgr.debugRegisterClientForTesting(clientB);
|
|
await db.insertWatchAction(
|
|
profileId: 'profile-a',
|
|
serverId: ServerId('plex-machine'),
|
|
clientScopeId: scopeA,
|
|
ratingKey: 'item-1',
|
|
actionType: OfflineActionType.watched.id,
|
|
);
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(clientB.watched, isEmpty);
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
});
|
|
});
|
|
|
|
group('Jellyfin scoped sync', () {
|
|
test('empty active scope falls back to the downloaded scope during client pre-bind', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
mgr.debugRegisterClientForTesting(
|
|
_ScopedRecordingMediaClient(
|
|
serverId: ServerId('jf-machine'),
|
|
backend: MediaBackend.jellyfin,
|
|
scopedServerId: '',
|
|
),
|
|
);
|
|
|
|
final returnedScope = await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1');
|
|
|
|
final queued = await db.getPendingWatchActions();
|
|
expect(returnedScope, 'jf-machine/user-a');
|
|
expect(queued.single.clientScopeId, 'jf-machine/user-a');
|
|
});
|
|
|
|
test('queues with downloaded Jellyfin source scope when no active client is registered', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1');
|
|
|
|
final queued = await db.getPendingWatchActions();
|
|
expect(queued.single.clientScopeId, 'jf-machine/user-a');
|
|
});
|
|
|
|
test('local status and resume offset use active scope over downloaded source scope', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final activeUserB = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-b'),
|
|
httpClient: MockClient((_) async => http.Response('{}', 200)),
|
|
);
|
|
addTearDown(activeUserB.close);
|
|
mgr.debugRegisterJellyfinClientForTesting(activeUserB);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
await db.upsertProgressAction(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
viewOffset: 5000,
|
|
duration: 100000,
|
|
shouldMarkWatched: false,
|
|
);
|
|
await Future<void>.delayed(const Duration(milliseconds: 2));
|
|
await db.upsertProgressAction(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-b',
|
|
ratingKey: 'item-1',
|
|
viewOffset: 90000,
|
|
duration: 100000,
|
|
shouldMarkWatched: true,
|
|
);
|
|
|
|
expect(await svc.getLocalWatchStatus('jf-machine:item-1'), isTrue);
|
|
expect(await svc.getLocalViewOffset('jf-machine:item-1'), isNull);
|
|
expect(await svc.getLocalWatchStatus('jf-machine:item-1', clientScopeId: 'jf-machine/user-a'), isNull);
|
|
expect(await svc.getLocalViewOffset('jf-machine:item-1', clientScopeId: 'jf-machine/user-a'), 5000);
|
|
});
|
|
|
|
test('queues with active Jellyfin user instead of downloaded source scope', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
|
|
final activeUserB = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-b'),
|
|
httpClient: MockClient((_) async => http.Response('{}', 200)),
|
|
);
|
|
addTearDown(activeUserB.close);
|
|
mgr.debugRegisterJellyfinClientForTesting(activeUserB);
|
|
|
|
final returnedScope = await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1');
|
|
|
|
final queued = await db.getPendingWatchActions();
|
|
expect(returnedScope, 'jf-machine/user-b');
|
|
expect(queued.single.clientScopeId, 'jf-machine/user-b');
|
|
});
|
|
|
|
test('replays through the queued Jellyfin user after active user changes', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
svc.setActiveProfileId('p1');
|
|
|
|
final pathsByUser = <String, List<String>>{'user-a': [], 'user-b': []};
|
|
|
|
JellyfinClient clientFor(String userId) {
|
|
return JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection(userId),
|
|
httpClient: MockClient((request) async {
|
|
pathsByUser[userId]!.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
if (request.method == 'GET' && request.url.path == '/Users/$userId/Items/item-1') {
|
|
return http.Response('{"Id":"item-1","Type":"Movie","Name":"Movie $userId"}', 200);
|
|
}
|
|
if (request.method == 'POST' && request.url.path == '/UserPlayedItems/item-1') {
|
|
return http.Response('', 204);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
}
|
|
|
|
final userA = clientFor('user-a');
|
|
final userB = clientFor('user-b');
|
|
addTearDown(userA.close);
|
|
addTearDown(userB.close);
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(userA);
|
|
await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1');
|
|
final queued = await db.getPendingWatchActions();
|
|
expect(queued.single.clientScopeId, 'jf-machine/user-a');
|
|
|
|
// User B becomes the active machine client. The queued action must
|
|
// still resolve the specific user A client from clientScopeId.
|
|
mgr.debugRegisterJellyfinClientForTesting(userB, online: false);
|
|
await svc.syncPendingItems();
|
|
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
expect(pathsByUser['user-a'], contains('POST /UserPlayedItems/item-1?userId=user-a'));
|
|
expect(pathsByUser['user-b'], isEmpty);
|
|
});
|
|
|
|
test('legacy Jellyfin rows without clientScopeId are not synced through the active server client', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final paths = <String>[];
|
|
final client = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-b'),
|
|
httpClient: MockClient((request) async {
|
|
paths.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
if (request.method == 'GET' && request.url.path == '/Users/user-b/Items/item-1') {
|
|
return http.Response('{"Id":"item-1","Type":"Movie","Name":"Movie"}', 200);
|
|
}
|
|
if (request.method == 'POST' && request.url.path == '/UserPlayedItems/item-1') {
|
|
return http.Response('', 204);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
addTearDown(client.close);
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(client);
|
|
await db.insertWatchAction(
|
|
serverId: ServerId('jf-machine'),
|
|
ratingKey: 'item-1',
|
|
actionType: OfflineActionType.watched.id,
|
|
);
|
|
|
|
await svc.syncPendingItems();
|
|
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
expect(paths, isEmpty);
|
|
});
|
|
|
|
test('legacy Jellyfin rows without clientScopeId do not borrow downloaded source scope during replay', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final pathsByUser = <String, List<String>>{'user-a': [], 'user-b': []};
|
|
|
|
JellyfinClient clientFor(String userId) {
|
|
return JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection(userId),
|
|
httpClient: MockClient((request) async {
|
|
pathsByUser[userId]!.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
if (request.method == 'GET' && request.url.path == '/Users/$userId/Items/item-1') {
|
|
return http.Response('{"Id":"item-1","Type":"Movie","Name":"Movie $userId"}', 200);
|
|
}
|
|
if (request.method == 'POST' && request.url.path == '/UserPlayedItems/item-1') {
|
|
return http.Response('', 204);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
}
|
|
|
|
final userA = clientFor('user-a');
|
|
final userB = clientFor('user-b');
|
|
addTearDown(userA.close);
|
|
addTearDown(userB.close);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
await db.insertWatchAction(
|
|
serverId: ServerId('jf-machine'),
|
|
ratingKey: 'item-1',
|
|
actionType: OfflineActionType.watched.id,
|
|
);
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(userA);
|
|
mgr.debugRegisterJellyfinClientForTesting(userB);
|
|
await svc.syncPendingItems();
|
|
|
|
expect(await svc.getPendingSyncCount(), 1);
|
|
expect(pathsByUser['user-a'], isEmpty);
|
|
expect(pathsByUser['user-b'], isEmpty);
|
|
});
|
|
|
|
test('watch-state pull uses active scope for shared movie downloads', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final pathsByUser = <String, List<String>>{'user-a': [], 'user-b': []};
|
|
|
|
JellyfinClient clientFor(String userId) {
|
|
return JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection(userId),
|
|
httpClient: MockClient((request) async {
|
|
pathsByUser[userId]!.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
if (request.method == 'GET' && request.url.path == '/Users/$userId/Items/item-1') {
|
|
return http.Response(
|
|
'{"Id":"item-1","Type":"Movie","Name":"Movie $userId","UserData":{"PlayCount":1,"Played":true}}',
|
|
200,
|
|
headers: {'content-type': 'application/json'},
|
|
);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
}
|
|
|
|
final userA = clientFor('user-a');
|
|
final userB = clientFor('user-b');
|
|
addTearDown(userA.close);
|
|
addTearDown(userB.close);
|
|
final events = <WatchStateEvent>[];
|
|
final sub = WatchStateNotifier().stream.listen(events.add);
|
|
addTearDown(sub.cancel);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
await db.addDownloadOwner(profileId: 'profile-b', globalKey: 'jf-machine:item-1');
|
|
svc.setActiveProfileId('profile-b');
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(userA);
|
|
mgr.debugRegisterJellyfinClientForTesting(userB);
|
|
await svc.syncWatchStatesFromServer();
|
|
await Future<void>.delayed(Duration.zero);
|
|
|
|
expect(pathsByUser['user-a'], isEmpty);
|
|
expect(pathsByUser['user-b']!.where((p) => p.startsWith('GET /Users/user-b/Items/item-1?')), isNotEmpty);
|
|
expect(events.single.cacheServerId, 'jf-machine/user-b');
|
|
});
|
|
|
|
test('watch-state pull does not treat Jellyfin PlayCount alone as watched', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final paths = <String>[];
|
|
final userB = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-b'),
|
|
httpClient: MockClient((request) async {
|
|
paths.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
if (request.method == 'GET' && request.url.path == '/Users/user-b/Items/item-1') {
|
|
return http.Response(
|
|
'{"Id":"item-1","Type":"Movie","Name":"Started Movie","UserData":{"PlayCount":1,"Played":false}}',
|
|
200,
|
|
headers: {'content-type': 'application/json'},
|
|
);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
addTearDown(userB.close);
|
|
final events = <WatchStateEvent>[];
|
|
final sub = WatchStateNotifier().stream.listen(events.add);
|
|
addTearDown(sub.cancel);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-b',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
await db.addDownloadOwner(profileId: 'profile-b', globalKey: 'jf-machine:item-1');
|
|
svc.setActiveProfileId('profile-b');
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(userB);
|
|
await svc.syncWatchStatesFromServer();
|
|
await Future<void>.delayed(Duration.zero);
|
|
|
|
expect(paths.where((p) => p.startsWith('GET /Users/user-b/Items/item-1?')), isNotEmpty);
|
|
expect(events, isEmpty);
|
|
});
|
|
|
|
test('watch-state pull uses active scope for shared episode season batches', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final pathsByUser = <String, List<String>>{'user-a': [], 'user-b': []};
|
|
|
|
JellyfinClient clientFor(String userId) {
|
|
return JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection(userId),
|
|
httpClient: MockClient((request) async {
|
|
pathsByUser[userId]!.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
if (request.method == 'GET' && request.url.path == '/Shows/season-1/Seasons') {
|
|
return http.Response('not found', 404);
|
|
}
|
|
if (request.method == 'GET' && request.url.path == '/Items') {
|
|
return http.Response(
|
|
'{"Items":[{"Id":"ep-1","Type":"Episode","Name":"Episode $userId","UserData":{"PlayCount":1,"Played":true}}]}',
|
|
200,
|
|
headers: {'content-type': 'application/json'},
|
|
);
|
|
}
|
|
if (request.method == 'GET' && request.url.path == '/Users/$userId/Items/ep-1') {
|
|
return http.Response(
|
|
'{"Id":"ep-1","Type":"Episode","Name":"Episode $userId","UserData":{"PlayCount":1,"Played":true}}',
|
|
200,
|
|
headers: {'content-type': 'application/json'},
|
|
);
|
|
}
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
}
|
|
|
|
final userA = clientFor('user-a');
|
|
final userB = clientFor('user-b');
|
|
addTearDown(userA.close);
|
|
addTearDown(userB.close);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'ep-1',
|
|
globalKey: 'jf-machine:ep-1',
|
|
type: 'episode',
|
|
parentRatingKey: 'season-1',
|
|
status: 3,
|
|
);
|
|
await db.addDownloadOwner(profileId: 'profile-b', globalKey: 'jf-machine:ep-1');
|
|
svc.setActiveProfileId('profile-b');
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(userA);
|
|
mgr.debugRegisterJellyfinClientForTesting(userB);
|
|
await svc.syncWatchStatesFromServer();
|
|
|
|
expect(pathsByUser['user-a'], isEmpty);
|
|
expect(pathsByUser['user-b']!.where((p) => p.startsWith('GET /Items?')), isNotEmpty);
|
|
});
|
|
|
|
test('watch-state pull ignores physical downloads not owned by active profile', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final paths = <String>[];
|
|
final userB = JellyfinClient.forTesting(
|
|
connection: _jellyfinConnection('user-b'),
|
|
httpClient: MockClient((request) async {
|
|
paths.add('${request.method} ${request.url.path}?${request.url.query}');
|
|
return http.Response('not found', 404);
|
|
}),
|
|
);
|
|
addTearDown(userB.close);
|
|
|
|
await db.insertDownload(
|
|
serverId: ServerId('jf-machine'),
|
|
clientScopeId: 'jf-machine/user-a',
|
|
ratingKey: 'item-1',
|
|
globalKey: 'jf-machine:item-1',
|
|
type: 'movie',
|
|
status: 3,
|
|
);
|
|
await db.addDownloadOwner(profileId: 'profile-a', globalKey: 'jf-machine:item-1');
|
|
svc.setActiveProfileId('profile-b');
|
|
|
|
mgr.debugRegisterJellyfinClientForTesting(userB);
|
|
await svc.syncWatchStatesFromServer();
|
|
|
|
expect(paths, isEmpty);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// clearAll
|
|
// ============================================================
|
|
|
|
group('clearAll', () {
|
|
test('removes every queued action and notifies listeners', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '1');
|
|
await svc.queueMarkUnwatched(serverId: ServerId('srv'), itemId: '2');
|
|
expect(await svc.getPendingSyncCount(), 2);
|
|
|
|
var notifications = 0;
|
|
svc.addListener(() => notifications++);
|
|
|
|
await svc.clearAll();
|
|
expect(await svc.getPendingSyncCount(), 0);
|
|
expect(notifications, 1);
|
|
});
|
|
});
|
|
|
|
// ============================================================
|
|
// startConnectivityMonitoring + dispose
|
|
// ============================================================
|
|
|
|
group('startConnectivityMonitoring + dispose', () {
|
|
test('attaches a listener to the source', () {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final source = _FakeOfflineModeSource();
|
|
expect(source.hasListeners, isFalse);
|
|
|
|
svc.startConnectivityMonitoring(source);
|
|
expect(source.hasListeners, isTrue);
|
|
|
|
svc.dispose();
|
|
// After dispose, the listener is removed.
|
|
expect(source.hasListeners, isFalse);
|
|
});
|
|
|
|
test('replacing the source detaches the prior listener', () {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
svc.dispose();
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
final first = _FakeOfflineModeSource();
|
|
final second = _FakeOfflineModeSource();
|
|
|
|
svc.startConnectivityMonitoring(first);
|
|
expect(first.hasListeners, isTrue);
|
|
expect(second.hasListeners, isFalse);
|
|
|
|
svc.startConnectivityMonitoring(second);
|
|
// First's listener was removed; second now has one.
|
|
expect(first.hasListeners, isFalse);
|
|
expect(second.hasListeners, isTrue);
|
|
});
|
|
|
|
test('dispose() before startConnectivityMonitoring is safe', () async {
|
|
final (svc: svc, db: db, mgr: mgr) = _makeService();
|
|
addTearDown(() async {
|
|
mgr.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
// Never called startConnectivityMonitoring → both fields are null.
|
|
expect(svc.dispose, returnsNormally);
|
|
});
|
|
});
|
|
}
|