edde746
35061f9f68
Jellyfin search rows do not expose their owning collection, so hidden libraries cannot be filtered after the response. Scope searches to visible libraries, stamp the returned rows, reuse the latest loaded views, propagate cancellation, and fail closed when views cannot be loaded. Keep full candidate budgets and split music libraries into parallel album, audio, and artist requests. Album requests disable UserData and use the existing album field set to avoid recursive per-album work from #1552; audio requests retain cheap leaf play state. close #1770
687 lines
27 KiB
Dart
687 lines
27 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;
|
|
final hubs = shouldUseGlobalHubs
|
|
? [
|
|
...await 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.
|
|
...await _fetchLibraryHubsForClient(
|
|
client,
|
|
limit: hubItemLimit,
|
|
hiddenLibraryKeys: hiddenLibraryKeys,
|
|
includePlaybackHubs: includePlaybackHubs,
|
|
libraries: serverLibraries ?? const [],
|
|
kinds: const {MediaKind.artist},
|
|
),
|
|
]
|
|
: 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();
|
|
|
|
const concurrency = 3;
|
|
final all = <MediaHub>[];
|
|
for (var start = 0; start < visible.length; start += concurrency) {
|
|
final batch = visible.skip(start).take(concurrency);
|
|
final results = await Future.wait(
|
|
batch.map((l) async {
|
|
try {
|
|
return await client.fetchLibraryHubs(
|
|
l.id,
|
|
libraryName: l.title,
|
|
limit: limit,
|
|
includePlaybackHubs: includePlaybackHubs,
|
|
libraryKind: l.kind,
|
|
);
|
|
} catch (e, st) {
|
|
appLogger.e('Failed to fetch library hubs for ${l.globalKey}', error: e, stackTrace: st);
|
|
return <MediaHub>[];
|
|
}
|
|
}),
|
|
);
|
|
for (final list in results) {
|
|
all.addAll(list);
|
|
}
|
|
}
|
|
return all;
|
|
}
|
|
|
|
/// 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;
|
|
}
|
|
}
|