The Discover tab fanned out its whole request set twice on every cold start and replayed slow rows on a shrinking timeout ladder, so a healthy remote server produced anywhere from 4s to 15s of loading. Measured against a remote Jellyfin server with four libraries, 24 interleaved cold-start samples per side: requests 19 -> 9 payload 219 KB -> 94 KB settled 5231ms -> 2502ms median, 13222ms -> 5927ms p95 Four independent causes: - Retry policy. `Client.send` resolves on response headers, so the connect budget covers the server's think time and a slow-but-alive query raises `connectionTimeout`. Replaying it made the server re-run the query with a shorter budget than the one it just missed; the `[10s, 8s, 5s]` ladder turned an 11s answer into an empty row after 23s. Hub surfaces now get one whole-request deadline, retry only immediate connection errors, and the deadline bounds the whole call including the request still in flight. - Request shape. `/Items/Latest` groups a TV library by series, so its rows are Series folder dtos and `RecursiveItemCount`/`ChildCount` cost a DB count each, per row. Hub rows now ask for `Overview` only; watch state survives because Jellyfin derives `UserData.Played` from `UnplayedItemCount` when the count fields are absent. `/Shows/NextUp` sends `NextUpDateCutoff` to bound the server's series-key scan, and `Thumb` leaves `EnableImageTypes` since nothing reads it. `UserData` and `PremiereDate` leave the browse set: neither is an `ItemFields` member, so the server dropped them anyway. - Fan-out. Per-library hubs ran in batches of three separated by a barrier, so one slow library stalled every library behind it. A sliding window keeps the same peak concurrency without head-of-line blocking. Concurrent `fetchLibraries` calls now share one `/Views` instead of racing two identical round trips, Plex's global and music hub legs start together, and Jellyfin gets Plex's pool tuning. - Duplicate pass. `DiscoverScreen.initState` starts a load and the online-entry hook asked for a full refresh on top of it, which `CoalescedLoadCoordinator` correctly queued as a trailing pass. The hook now calls `primeRefresh`, which rides along with a load already in flight; profile switches still go through `fullRefresh`. Refs #1784
701 lines
28 KiB
Dart
701 lines
28 KiB
Dart
import 'dart:async';
|
|
import '../media/ids.dart';
|
|
|
|
import '../media/media_hub.dart';
|
|
import '../media/media_item.dart';
|
|
import '../media/media_item_merge.dart';
|
|
import '../media/media_kind.dart';
|
|
import '../media/media_library.dart';
|
|
import '../media/media_server_client.dart';
|
|
import '../exceptions/media_server_exceptions.dart';
|
|
import '../utils/app_logger.dart';
|
|
import '../utils/external_ids.dart';
|
|
import '../utils/global_key_utils.dart';
|
|
import '../utils/search_relevance.dart';
|
|
import '../utils/media_server_http_client.dart';
|
|
import 'local_playback_history.dart';
|
|
import 'multi_server_manager.dart';
|
|
|
|
typedef OnDeckAggregationResult = ({
|
|
List<MediaItem> items,
|
|
Set<String> succeededServerIds,
|
|
Set<String> cancelledServerIds,
|
|
});
|
|
typedef HubAggregationResult = ({List<MediaHub> hubs, Set<String> succeededServerIds, Set<String> cancelledServerIds});
|
|
typedef LibraryAggregationResult = ({
|
|
List<MediaLibrary> libraries,
|
|
Set<String> succeededServerIds,
|
|
Set<String> cancelledServerIds,
|
|
});
|
|
typedef SearchAggregationResult = ({
|
|
List<MediaItem> items,
|
|
Set<String> succeededServerIds,
|
|
Set<String> cancelledServerIds,
|
|
Set<String> failedServerIds,
|
|
});
|
|
typedef _FanOutResult<T> = ({
|
|
List<T> items,
|
|
Set<String> succeededServerIds,
|
|
Set<String> cancelledServerIds,
|
|
Set<String> failedServerIds,
|
|
});
|
|
|
|
/// Whether [error] is a client-side abort (client teardown mid-request)
|
|
/// rather than a genuine server failure. Aggregation reports these servers
|
|
/// in `cancelledServerIds` so callers can tell a *disrupted* pass — whose
|
|
/// results say nothing about actual content — from a settled failure.
|
|
bool _isCancellation(Object error) => error is MediaServerHttpException && error.isCancellation;
|
|
|
|
Map<String, int> _searchKindCounts(Iterable<MediaItem> items) {
|
|
final counts = <String, int>{};
|
|
for (final item in items) {
|
|
counts.update(item.kind.name, (count) => count + 1, ifAbsent: () => 1);
|
|
}
|
|
return counts;
|
|
}
|
|
|
|
/// Drop items belonging to a hidden library.
|
|
///
|
|
/// Items the backend could not attribute to a library ([MediaItem.libraryGlobalKey]
|
|
/// is null) are kept: Plex search and `/library/shared/all` return shared and
|
|
/// external rows that have no local section, and those are not something the
|
|
/// user hid.
|
|
List<MediaItem> _withoutHiddenLibraries(List<MediaItem> items, Set<String>? hiddenLibraryKeys) {
|
|
if (hiddenLibraryKeys == null || hiddenLibraryKeys.isEmpty) return items;
|
|
return items.where((item) {
|
|
final libraryKey = item.libraryGlobalKey;
|
|
return libraryKey == null || !hiddenLibraryKeys.contains(libraryKey);
|
|
}).toList();
|
|
}
|
|
|
|
/// The server-local library ids [serverId] owns within [hiddenLibraryKeys],
|
|
/// which hold cross-server `serverId:libraryId` global keys. Backends that
|
|
/// cannot attribute a search hit to a library need these to scope the request.
|
|
Set<String> _hiddenLibraryIdsOn(String serverId, Set<String>? hiddenLibraryKeys) {
|
|
if (hiddenLibraryKeys == null || hiddenLibraryKeys.isEmpty) return const {};
|
|
final ids = <String>{};
|
|
for (final key in hiddenLibraryKeys) {
|
|
final parsed = parseGlobalKey(key);
|
|
if (parsed != null && parsed.serverId == serverId) ids.add(parsed.ratingKey);
|
|
}
|
|
return ids;
|
|
}
|
|
|
|
/// Cross-server aggregation: fans calls out to every online client and
|
|
/// merges the results. Single-server operations now go through the
|
|
/// [MediaServerClient] interface directly (resolved via
|
|
/// [ProviderExtensions.tryGetMediaClientForServer] etc.), so this service
|
|
/// only owns the genuinely multi-server flows: home/discover hubs, on-deck,
|
|
/// search, and the global library list.
|
|
class DataAggregationService {
|
|
final MultiServerManager _serverManager;
|
|
|
|
DataAggregationService(this._serverManager);
|
|
|
|
/// Online clients, optionally restricted to [serverIds] — delta refreshes
|
|
/// fan out to newly-online servers only.
|
|
Map<String, MediaServerClient> _clientsFor(Set<String>? serverIds) {
|
|
final clients = _serverManager.onlineClients;
|
|
if (serverIds == null) return clients;
|
|
return {
|
|
for (final entry in clients.entries)
|
|
if (serverIds.contains(entry.key)) entry.key: entry.value,
|
|
};
|
|
}
|
|
|
|
/// Run [fetch] against every client in [clients] and concatenate the results
|
|
/// in client order. A per-server failure is swallowed — logged with
|
|
/// [failureMessage] and contributing nothing — so one bad server cannot sink
|
|
/// the pass; that server is simply absent from `succeededServerIds` and lands
|
|
/// in `failedServerIds`. A client-side abort is *not* a failure: it lands in
|
|
/// `cancelledServerIds` and is logged at debug level, so callers can tell a
|
|
/// disrupted pass from a settled one and torn-down requests do not spam the
|
|
/// error log.
|
|
Future<_FanOutResult<T>> _fanOut<T>(
|
|
Map<String, MediaServerClient> clients, {
|
|
required String Function(String serverId) failureMessage,
|
|
required Future<List<T>> Function(String serverId, MediaServerClient client) fetch,
|
|
}) async {
|
|
final cancelledServerIds = <String>{};
|
|
final failedServerIds = <String>{};
|
|
final futures = clients.entries.map((entry) async {
|
|
try {
|
|
return (serverId: entry.key, items: await fetch(entry.key, entry.value));
|
|
} catch (e, stackTrace) {
|
|
if (_isCancellation(e)) {
|
|
cancelledServerIds.add(entry.key);
|
|
appLogger.d('Cancelled (client abort): ${failureMessage(entry.key)}');
|
|
} else {
|
|
failedServerIds.add(entry.key);
|
|
appLogger.e(failureMessage(entry.key), error: e, stackTrace: stackTrace);
|
|
}
|
|
return (serverId: null, items: <T>[]);
|
|
}
|
|
});
|
|
final results = await Future.wait(futures);
|
|
return (
|
|
items: [for (final result in results) ...result.items],
|
|
succeededServerIds: {
|
|
for (final result in results)
|
|
if (result.serverId != null) result.serverId!,
|
|
},
|
|
cancelledServerIds: cancelledServerIds,
|
|
failedServerIds: failedServerIds,
|
|
);
|
|
}
|
|
|
|
/// Fetch libraries from all online clients regardless of backend, returning
|
|
/// the merged neutral [MediaLibrary]s alongside the ids of the servers whose
|
|
/// fetch actually succeeded. [serverIds] restricts the fan-out to those
|
|
/// servers.
|
|
///
|
|
/// A per-server `fetchLibraries()` failure is swallowed (that server simply
|
|
/// contributes no libraries) so one unreachable server doesn't sink the whole
|
|
/// list. [succeededServerIds] lets callers tell a *failed* fetch apart from a
|
|
/// server that genuinely has no libraries — both contribute nothing, so
|
|
/// conflating them would let a transient failure be cached as "loaded" and
|
|
/// never retried. Servers whose fetch was aborted client-side land in
|
|
/// `cancelledServerIds` — a disrupted pass, unlike a settled failure, must
|
|
/// never be committed as authoritative.
|
|
Future<LibraryAggregationResult> getMediaLibrariesFromAllServers({Set<String>? serverIds}) async {
|
|
final clients = _clientsFor(serverIds);
|
|
if (clients.isEmpty) {
|
|
appLogger.w('No online servers available for fetching libraries (neutral)');
|
|
return (
|
|
libraries: const <MediaLibrary>[],
|
|
succeededServerIds: const <String>{},
|
|
cancelledServerIds: const <String>{},
|
|
);
|
|
}
|
|
final fetched = await _fanOut<MediaLibrary>(
|
|
clients,
|
|
failureMessage: (serverId) => 'Failed neutral library fetch from $serverId',
|
|
fetch: (_, client) => client.fetchLibraries(),
|
|
);
|
|
return (
|
|
libraries: fetched.items,
|
|
succeededServerIds: fetched.succeededServerIds,
|
|
cancelledServerIds: fetched.cancelledServerIds,
|
|
);
|
|
}
|
|
|
|
/// Fetch "On Deck" (Continue Watching) from all servers and merge by recency.
|
|
/// Items are tagged with server info by the underlying client. Returns
|
|
/// neutral [MediaItem]s plus the ids of servers whose fetch succeeded.
|
|
/// [serverIds] restricts the fan-out to those servers.
|
|
Future<OnDeckAggregationResult> getOnDeckFromAllServers({
|
|
int? limit,
|
|
Set<String>? hiddenLibraryKeys,
|
|
Set<String>? serverIds,
|
|
}) async {
|
|
final clients = _clientsFor(serverIds);
|
|
if (clients.isEmpty) {
|
|
appLogger.w('No online servers available for fetching on deck');
|
|
return (items: const <MediaItem>[], succeededServerIds: const <String>{}, cancelledServerIds: const <String>{});
|
|
}
|
|
|
|
final fetched = await _fanOut<MediaItem>(
|
|
clients,
|
|
failureMessage: (serverId) => 'Failed on-deck fetch from $serverId',
|
|
fetch: (_, client) => client.fetchContinueWatching(count: limit),
|
|
);
|
|
// Filter out items from hidden libraries
|
|
var filteredOnDeck = _withoutHiddenLibraries(fetched.items, hiddenLibraryKeys);
|
|
|
|
// Sort by most recently viewed, falling back to addedAt for unwatched items.
|
|
// Same key as JellyfinClient's continue-watching merge (MediaItem.recencySortKey)
|
|
// so per-server and cross-server ordering can't drift apart.
|
|
filteredOnDeck.sort((a, b) => b.recencySortKey.compareTo(a.recencySortKey));
|
|
|
|
filteredOnDeck = await _deduplicateContinueWatching(filteredOnDeck);
|
|
|
|
// Apply limit if specified
|
|
final items = limit != null && limit < filteredOnDeck.length ? filteredOnDeck.sublist(0, limit) : filteredOnDeck;
|
|
|
|
appLogger.i('Fetched ${items.length} on deck items from all servers');
|
|
|
|
return (
|
|
items: items,
|
|
succeededServerIds: fetched.succeededServerIds,
|
|
cancelledServerIds: fetched.cancelledServerIds,
|
|
);
|
|
}
|
|
|
|
/// Merge an [existing] Continue Watching list with [fresh] rows from
|
|
/// newly-online servers: same recency ordering and cross-server identity
|
|
/// dedup as [getOnDeckFromAllServers], applied to the union.
|
|
Future<List<MediaItem>> mergeContinueWatching(List<MediaItem> existing, List<MediaItem> fresh, {int? limit}) async {
|
|
final combined = [...existing, ...fresh]..sort((a, b) => b.recencySortKey.compareTo(a.recencySortKey));
|
|
final deduped = await _deduplicateContinueWatching(combined);
|
|
return limit != null && limit < deduped.length ? deduped.sublist(0, limit) : deduped;
|
|
}
|
|
|
|
Future<List<MediaItem>> _deduplicateContinueWatching(List<MediaItem> items) async {
|
|
if (items.length < 2) return items;
|
|
|
|
final bucketCounts = <String, int>{};
|
|
for (final item in items) {
|
|
final bucket = _continueWatchingTitleBucket(item);
|
|
if (bucket == null) continue;
|
|
bucketCounts[bucket] = (bucketCounts[bucket] ?? 0) + 1;
|
|
}
|
|
|
|
final duplicateBuckets = {
|
|
for (final entry in bucketCounts.entries)
|
|
if (entry.value > 1) entry.key,
|
|
};
|
|
if (duplicateBuckets.isEmpty) return items;
|
|
|
|
final externalIdLoads = <String, Future<ExternalIds>>{};
|
|
final identityKeysByIndex = <int, Set<String>>{};
|
|
final identityKeyLoads = <Future<void>>[];
|
|
for (var i = 0; i < items.length; i++) {
|
|
if (!duplicateBuckets.contains(_continueWatchingTitleBucket(items[i]))) continue;
|
|
final index = i;
|
|
identityKeyLoads.add(
|
|
_continueWatchingIdentityKeys(items[index], externalIdLoads).then((keys) => identityKeysByIndex[index] = keys),
|
|
);
|
|
}
|
|
await Future.wait(identityKeyLoads);
|
|
|
|
// Group duplicates instead of greedily dropping them: the first item to
|
|
// claim an identity key anchors the group and holds its shelf slot;
|
|
// later items sharing a claimed key join as members without claiming
|
|
// their own keys (same transitive semantics as the old drop). Each slot
|
|
// then shows the member the user most recently played on this device —
|
|
// servers sync watch state across guid-linked siblings, so their
|
|
// lastViewedAt ties and can't tell the 4K copy from the 1080p one
|
|
// (#1492). Without local history the anchor (recency order) stands.
|
|
final keyToGroup = <String, int>{};
|
|
final groups = <List<MediaItem>>[];
|
|
final groupSlots = <int, int>{};
|
|
final result = <MediaItem>[];
|
|
for (var i = 0; i < items.length; i++) {
|
|
final item = items[i];
|
|
if (!duplicateBuckets.contains(_continueWatchingTitleBucket(item))) {
|
|
result.add(item);
|
|
continue;
|
|
}
|
|
|
|
final identityKeys = identityKeysByIndex[i] ?? const <String>{};
|
|
if (identityKeys.isEmpty) {
|
|
result.add(item);
|
|
continue;
|
|
}
|
|
|
|
var joined = false;
|
|
for (final key in identityKeys) {
|
|
final groupIndex = keyToGroup[key];
|
|
if (groupIndex != null) {
|
|
groups[groupIndex].add(item);
|
|
joined = true;
|
|
break;
|
|
}
|
|
}
|
|
if (joined) continue;
|
|
|
|
final groupIndex = groups.length;
|
|
groups.add([item]);
|
|
for (final key in identityKeys) {
|
|
keyToGroup[key] = groupIndex;
|
|
}
|
|
groupSlots[result.length] = groupIndex;
|
|
result.add(item);
|
|
}
|
|
|
|
if (groupSlots.isEmpty) return result;
|
|
final lastPlayed = await LocalPlaybackHistory.snapshot();
|
|
for (final slot in groupSlots.entries) {
|
|
final members = groups[slot.value];
|
|
if (members.length > 1) {
|
|
result[slot.key] = _preferLocallyLastPlayed(members, lastPlayed);
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
/// The duplicate-group member most recently played on this device (by item
|
|
/// or series key), or the anchor — `members.first`, the group's most recent
|
|
/// item by [MediaItem.recencySortKey] — when the local history has nothing
|
|
/// newer to say.
|
|
MediaItem _preferLocallyLastPlayed(List<MediaItem> members, Map<String, int> lastPlayed) {
|
|
var winner = members.first;
|
|
var winnerLastPlayedAt = 0;
|
|
for (final member in members) {
|
|
final itemTs = lastPlayed[member.globalKey] ?? 0;
|
|
final seriesKey = member.seriesGlobalKey;
|
|
final seriesTs = seriesKey != null ? (lastPlayed[seriesKey] ?? 0) : 0;
|
|
final lastPlayedAt = itemTs > seriesTs ? itemTs : seriesTs;
|
|
if (lastPlayedAt > winnerLastPlayedAt) {
|
|
winner = member;
|
|
winnerLastPlayedAt = lastPlayedAt;
|
|
}
|
|
}
|
|
return winner;
|
|
}
|
|
|
|
String? _continueWatchingTitleBucket(MediaItem item) {
|
|
final scope = _continueWatchingIdentityScope(item);
|
|
if (scope == null) return null;
|
|
|
|
final title = switch (item.kind) {
|
|
MediaKind.episode || MediaKind.season => item.grandparentTitle ?? item.parentTitle ?? item.title,
|
|
_ => item.title,
|
|
};
|
|
final normalized = title?.trim().toLowerCase().replaceAll(RegExp(r'\s+'), ' ');
|
|
if (normalized == null || normalized.isEmpty) return null;
|
|
return '$scope:$normalized';
|
|
}
|
|
|
|
Future<Set<String>> _continueWatchingIdentityKeys(
|
|
MediaItem item,
|
|
Map<String, Future<ExternalIds>> externalIdLoads,
|
|
) async {
|
|
final scope = _continueWatchingIdentityScope(item);
|
|
if (scope == null) return const {};
|
|
|
|
final keys = <String>{};
|
|
final serverId = item.serverId;
|
|
final targetId = _continueWatchingIdentityTargetId(item);
|
|
final client = serverId == null ? null : _serverManager.getClient(ServerId(serverId));
|
|
|
|
if (client != null && targetId != null && targetId.isNotEmpty) {
|
|
try {
|
|
final cacheKey = buildGlobalKey(ServerId(serverId!), targetId);
|
|
final externalIds = await externalIdLoads.putIfAbsent(cacheKey, () => client.fetchExternalIds(targetId));
|
|
_addExternalIdentityKeys(keys, scope, externalIds);
|
|
} catch (e, stackTrace) {
|
|
appLogger.d(
|
|
'Failed to resolve Continue Watching identity for ${item.globalKey}',
|
|
error: e,
|
|
stackTrace: stackTrace,
|
|
);
|
|
}
|
|
}
|
|
|
|
final stableGuid = _stableMediaGuid(item.guid);
|
|
if (stableGuid != null) {
|
|
final guidScope = item.kind == MediaKind.episode ? 'episode' : scope;
|
|
keys.add('$guidScope:guid:$stableGuid');
|
|
}
|
|
|
|
return keys;
|
|
}
|
|
|
|
String? _continueWatchingIdentityScope(MediaItem item) {
|
|
return switch (item.kind) {
|
|
MediaKind.episode || MediaKind.season || MediaKind.show => 'show',
|
|
MediaKind.movie => 'movie',
|
|
_ => null,
|
|
};
|
|
}
|
|
|
|
String? _continueWatchingIdentityTargetId(MediaItem item) {
|
|
return switch (item.kind) {
|
|
MediaKind.episode => item.grandparentId,
|
|
MediaKind.season => item.grandparentId ?? item.parentId,
|
|
MediaKind.show || MediaKind.movie => item.id,
|
|
_ => null,
|
|
};
|
|
}
|
|
|
|
void _addExternalIdentityKeys(Set<String> keys, String scope, ExternalIds externalIds) {
|
|
final imdb = externalIds.imdb?.trim().toLowerCase();
|
|
if (imdb != null && imdb.isNotEmpty) keys.add('$scope:imdb:$imdb');
|
|
final tmdb = externalIds.tmdb;
|
|
if (tmdb != null) keys.add('$scope:tmdb:$tmdb');
|
|
final tvdb = externalIds.tvdb;
|
|
if (tvdb != null) keys.add('$scope:tvdb:$tvdb');
|
|
}
|
|
|
|
String? _stableMediaGuid(String? guid) {
|
|
final value = guid?.trim();
|
|
if (value == null || value.isEmpty) return null;
|
|
if (!value.contains('://')) return null;
|
|
if (value.contains('agents.none://')) return null;
|
|
return value.toLowerCase();
|
|
}
|
|
|
|
/// Fetch recommendation hubs from all servers as neutral [MediaHub]s.
|
|
/// When useGlobalHubs is true (default), rich-hub backends use their true
|
|
/// home page hubs (Plex's promoted/global hub endpoint).
|
|
/// Backends without rich home hubs fall back to per-library hubs so one
|
|
/// capped "Latest" response cannot hide whole library types.
|
|
/// [serverIds] restricts the fan-out (including the library prefetch) to
|
|
/// those servers. Returns the ids of servers whose hub fetch succeeded so
|
|
/// callers do not cache transient per-server failures as loaded.
|
|
Future<HubAggregationResult> getHubsFromAllServers({
|
|
int? limit,
|
|
Set<String>? hiddenLibraryKeys,
|
|
bool useGlobalHubs = true,
|
|
bool includePlaybackHubs = true,
|
|
Set<String>? serverIds,
|
|
}) async {
|
|
final clients = _clientsFor(serverIds);
|
|
if (clients.isEmpty) {
|
|
appLogger.w('No online servers available for fetching hubs');
|
|
return (hubs: const <MediaHub>[], succeededServerIds: const <String>{}, cancelledServerIds: const <String>{});
|
|
}
|
|
|
|
// Home layout needs the library list for every client: fallback backends
|
|
// build all their rows from per-library hubs, and rich-hub backends
|
|
// (Plex) need it to detect visible music libraries, whose hubs the
|
|
// global-hub endpoint excludes. One `fetchLibraries` per server, served
|
|
// from the per-backend API cache when warm.
|
|
final libraries = useGlobalHubs
|
|
? _groupLibrariesByServer((await getMediaLibrariesFromAllServers(serverIds: serverIds)).libraries)
|
|
: null;
|
|
|
|
final fetched = await _fanOut<MediaHub>(
|
|
clients,
|
|
failureMessage: (serverId) => 'Failed to fetch hubs from server $serverId',
|
|
fetch: (serverId, client) async {
|
|
final serverLibraries = libraries?[serverId];
|
|
final shouldUseGlobalHubs = useGlobalHubs && client.capabilities.richHubs;
|
|
final hubItemLimit = limit ?? defaultHubPreviewLimit;
|
|
List<MediaHub> hubs;
|
|
if (shouldUseGlobalHubs) {
|
|
// Both legs are independent, so start them before awaiting either.
|
|
// Spreading `...await a, ...await b` into one list literal evaluates
|
|
// them in order, which serialised the music rows behind the global
|
|
// hub round trip.
|
|
final globalFuture = client.fetchGlobalHubs(limit: hubItemLimit, includePlaybackHubs: includePlaybackHubs);
|
|
// Plex's promoted/global hub endpoint never includes music
|
|
// libraries — append their per-library hubs so music rows
|
|
// reach home. No-op (zero extra calls) without a visible
|
|
// music library.
|
|
final musicFuture = _fetchLibraryHubsForClient(
|
|
client,
|
|
limit: hubItemLimit,
|
|
hiddenLibraryKeys: hiddenLibraryKeys,
|
|
includePlaybackHubs: includePlaybackHubs,
|
|
libraries: serverLibraries ?? const [],
|
|
kinds: const {MediaKind.artist},
|
|
);
|
|
hubs = [...await globalFuture, ...await musicFuture];
|
|
} else {
|
|
hubs = await _fetchLibraryHubsForClient(
|
|
client,
|
|
limit: hubItemLimit,
|
|
hiddenLibraryKeys: hiddenLibraryKeys,
|
|
includePlaybackHubs: includePlaybackHubs,
|
|
libraries: useGlobalHubs ? serverLibraries : null,
|
|
);
|
|
}
|
|
return _postProcessHubs(hubs, serverId: ServerId(serverId), hiddenLibraryKeys: hiddenLibraryKeys);
|
|
},
|
|
);
|
|
|
|
final all = fetched.items;
|
|
final hubs = limit != null && limit < all.length ? all.sublist(0, limit) : all;
|
|
return (hubs: hubs, succeededServerIds: fetched.succeededServerIds, cancelledServerIds: fetched.cancelledServerIds);
|
|
}
|
|
|
|
/// Per-library hub fetch for a single client. Filters to visible libraries
|
|
/// of [kinds] (movie/show/clip/artist by default — clip covers Jellyfin
|
|
/// musicvideos/homevideos, #1476; artist brings music rows to home) and
|
|
/// concatenates the results. The rich-hub music append passes
|
|
/// `{MediaKind.artist}` to fetch only what the global endpoint misses.
|
|
Future<List<MediaHub>> _fetchLibraryHubsForClient(
|
|
MediaServerClient client, {
|
|
required int limit,
|
|
Set<String>? hiddenLibraryKeys,
|
|
required bool includePlaybackHubs,
|
|
List<MediaLibrary>? libraries,
|
|
Set<MediaKind> kinds = const {MediaKind.movie, MediaKind.show, MediaKind.clip, MediaKind.artist},
|
|
}) async {
|
|
final libs = libraries ?? await client.fetchLibraries();
|
|
final visible = libs.where((l) {
|
|
if (!kinds.contains(l.kind)) return false;
|
|
if (l.hidden) return false;
|
|
if (hiddenLibraryKeys != null && hiddenLibraryKeys.contains(l.globalKey)) return false;
|
|
return true;
|
|
}).toList();
|
|
|
|
// Sliding window rather than batches of three separated by a barrier: a
|
|
// batch waits for its slowest member before the next one starts, so wall
|
|
// time was the sum of per-batch maxima and one slow library stalled every
|
|
// library queued behind it (#1784). Starting the next request the moment
|
|
// any slot frees keeps the same peak concurrency with no head-of-line
|
|
// blocking. Results are written back by index so hub order stays the
|
|
// library order regardless of completion order.
|
|
const concurrency = 3;
|
|
final results = List<List<MediaHub>>.filled(visible.length, const []);
|
|
var next = 0;
|
|
|
|
Future<void> worker() async {
|
|
while (true) {
|
|
final index = next++;
|
|
if (index >= visible.length) return;
|
|
final library = visible[index];
|
|
try {
|
|
results[index] = await client.fetchLibraryHubs(
|
|
library.id,
|
|
libraryName: library.title,
|
|
limit: limit,
|
|
includePlaybackHubs: includePlaybackHubs,
|
|
libraryKind: library.kind,
|
|
);
|
|
} catch (e, st) {
|
|
appLogger.e('Failed to fetch library hubs for ${library.globalKey}', error: e, stackTrace: st);
|
|
}
|
|
}
|
|
}
|
|
|
|
await Future.wait([for (var i = 0; i < concurrency && i < visible.length; i++) worker()]);
|
|
|
|
return [for (final list in results) ...list];
|
|
}
|
|
|
|
/// Filter hidden-library items and drop empty hubs.
|
|
List<MediaHub> _postProcessHubs(List<MediaHub> hubs, {required ServerId serverId, Set<String>? hiddenLibraryKeys}) {
|
|
var filtered = hubs;
|
|
if (hiddenLibraryKeys != null && hiddenLibraryKeys.isNotEmpty) {
|
|
filtered = filtered
|
|
.map((hub) {
|
|
final filteredItems = hub.items.where((item) {
|
|
final libraryId = item.libraryId;
|
|
if (libraryId == null) return true;
|
|
final globalKey = buildGlobalKey(ServerId(serverId), libraryId);
|
|
return !hiddenLibraryKeys.contains(globalKey);
|
|
}).toList();
|
|
if (filteredItems.isEmpty) return null;
|
|
return hub.copyWith(items: filteredItems, size: filteredItems.length);
|
|
})
|
|
.whereType<MediaHub>()
|
|
.toList();
|
|
}
|
|
return filtered;
|
|
}
|
|
|
|
/// Search across all online servers (Plex + Jellyfin). Per-server outcomes
|
|
/// distinguish authoritative empty results from failed or cancelled legs.
|
|
///
|
|
/// [hiddenLibraryKeys] excludes results the user has hidden, matching every
|
|
/// other aggregated surface. Backends whose search rows carry no library id
|
|
/// cannot be filtered here; they must scope the search server-side instead.
|
|
Future<SearchAggregationResult> searchAcrossServers(
|
|
String query, {
|
|
int? limit,
|
|
Set<String>? hiddenLibraryKeys,
|
|
AbortController? abort,
|
|
}) async {
|
|
if (query.trim().isEmpty) {
|
|
return (
|
|
items: const <MediaItem>[],
|
|
succeededServerIds: const <String>{},
|
|
cancelledServerIds: const <String>{},
|
|
failedServerIds: const <String>{},
|
|
);
|
|
}
|
|
|
|
abort?.throwIfAborted();
|
|
final clients = _serverManager.onlineClients;
|
|
if (clients.isEmpty) {
|
|
return (
|
|
items: const <MediaItem>[],
|
|
succeededServerIds: const <String>{},
|
|
cancelledServerIds: const <String>{},
|
|
failedServerIds: const <String>{},
|
|
);
|
|
}
|
|
|
|
final resultLimit = limit ?? defaultMediaSearchLimit;
|
|
final fetchLimit = resultLimit < defaultMediaSearchLimit ? defaultMediaSearchLimit : resultLimit;
|
|
|
|
final fetched = await _fanOut<MediaItem>(
|
|
clients,
|
|
failureMessage: (serverId) => 'Search failed on $serverId',
|
|
fetch: (serverId, client) async {
|
|
final stopwatch = Stopwatch()..start();
|
|
final items = await client.searchItems(
|
|
query,
|
|
limit: fetchLimit,
|
|
abort: abort,
|
|
excludedLibraryIds: _hiddenLibraryIdsOn(serverId, hiddenLibraryKeys),
|
|
);
|
|
appLogger.i(
|
|
'Search completed on $serverId in ${stopwatch.elapsedMilliseconds}ms: '
|
|
'${items.length} results ${_searchKindCounts(items)}',
|
|
);
|
|
return items;
|
|
},
|
|
);
|
|
abort?.throwIfAborted();
|
|
// Before ranking, so hidden results cannot spend the `resultLimit` budget
|
|
// and silently shrink what the user sees.
|
|
final visible = _withoutHiddenLibraries(fetched.items, hiddenLibraryKeys);
|
|
final items = rankMediaSearchResults(visible, query, limit: resultLimit);
|
|
|
|
appLogger.i(
|
|
'Search aggregation completed: ${items.length} results '
|
|
'(${fetched.succeededServerIds.length} succeeded, ${fetched.cancelledServerIds.length} cancelled, '
|
|
'${fetched.failedServerIds.length} failed) ${_searchKindCounts(items)}',
|
|
);
|
|
|
|
return (
|
|
items: items,
|
|
succeededServerIds: fetched.succeededServerIds,
|
|
cancelledServerIds: fetched.cancelledServerIds,
|
|
failedServerIds: fetched.failedServerIds,
|
|
);
|
|
}
|
|
|
|
/// Reverse external-id lookup fanned out to every online server (see
|
|
/// [MediaServerClient.findByExternalIds]). One request wave per tap on an
|
|
/// Explore catalog item; per-server failures are logged and skipped.
|
|
///
|
|
/// Every server contributes every copy it holds, not one apiece: the same
|
|
/// movie routinely sits in a 4K library and an HD library on one server
|
|
/// (#1754). Results are deduped by global key and ordered best-first with
|
|
/// [compareLibraryCopies] so the chooser is stable across repeated passes.
|
|
///
|
|
/// Because per-server failures are dropped here, a caller holding earlier
|
|
/// results must merge rather than replace (see [mergeLibraryCopies]) — a
|
|
/// degraded wave is not evidence that a copy went away.
|
|
Future<List<MediaItem>> findByExternalIdsAcrossServers(
|
|
ExternalIds ids, {
|
|
required MediaKind kind,
|
|
List<String> titles = const [],
|
|
int? year,
|
|
String? plexGuid,
|
|
ExternalSeasonRef? season,
|
|
}) async {
|
|
if (!ids.hasAny && plexGuid == null) return [];
|
|
final clients = _serverManager.onlineClients;
|
|
if (clients.isEmpty) return [];
|
|
|
|
final futures = clients.entries.map((entry) async {
|
|
try {
|
|
return await entry.value.findByExternalIds(
|
|
ids,
|
|
kind: kind,
|
|
titles: titles,
|
|
year: year,
|
|
plexGuid: plexGuid,
|
|
season: season,
|
|
);
|
|
} catch (e, st) {
|
|
appLogger.w('External-id lookup failed on ${entry.key}', error: e, stackTrace: st);
|
|
return const <MediaItem>[];
|
|
}
|
|
});
|
|
|
|
return mergeLibraryCopies(const [], (await Future.wait(futures)).expand((items) => items));
|
|
}
|
|
|
|
/// Group libraries by server (internal aggregation helper).
|
|
Map<String, List<MediaLibrary>> _groupLibrariesByServer(List<MediaLibrary> libraries) {
|
|
final grouped = <String, List<MediaLibrary>>{};
|
|
|
|
for (final library in libraries) {
|
|
final serverId = library.serverId;
|
|
if (serverId != null) {
|
|
grouped.putIfAbsent(serverId, () => []).add(library);
|
|
}
|
|
}
|
|
|
|
return grouped;
|
|
}
|
|
}
|