import 'dart:async'; import '../../../shared/utils/app_logger.dart'; import '../../contracts/contracts.dart'; import '../ports/emby_gateway.dart'; import '../ports/server_repository.dart'; import '../ports/session_repository.dart'; import 'usecase_helpers.dart'; const _ucTag = 'UseCase'; Stream _fanOutPerServer({ required SessionRepository sessionRepository, required ServerRepository serverRepository, required Future?> Function(AuthedRequestContext context) fetchItems, required String logTag, }) { final controller = StreamController(); () async { try { final allSessions = await sessionRepository.listAll(); final allServers = await serverRepository.list(); final serverMap = {for (final s in allServers) s.id: s}; final futures = allSessions.map((session) async { final server = serverMap[session.serverId]; if (server == null) return; final context = AuthedRequestContext( baseUrl: server.baseUrl, token: session.token, userId: session.userId, ); try { final items = await fetchItems(context); if (items == null || items.isEmpty) return; controller.add( GlobalSearchServerResult( serverId: server.id, serverName: server.name, serverUrl: server.baseUrl, token: session.token, userId: session.userId, items: items, ), ); } catch (e) { AppLogger.warn(_ucTag, '$logTag on ${server.name} failed', e); } }); await Future.wait(futures); } catch (e) { controller.addError(e); } finally { controller.close(); } }(); return controller.stream; } class GetLibraryViewsUseCase extends AuthedUseCase { GetLibraryViewsUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Future execute() async { return embyGateway.getUserViews(await ctx()); } } class GetLibraryResumeUseCase extends AuthedUseCase { GetLibraryResumeUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Future execute() async { return embyGateway.getResumeItems(await ctx(), limit: 20); } } class GetLibraryLatestUseCase extends AuthedUseCase { GetLibraryLatestUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Future execute( LibraryLatestReq input, { CancellationToken? cancelToken, }) async { return embyGateway.getLatestItems( await ctx(), input.parentId, limit: input.limit ?? 16, cancelToken: cancelToken, ); } } class GetLibraryItemsUseCase extends AuthedUseCase { GetLibraryItemsUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Future execute(LibraryItemsReq input) async { final items = await embyGateway.getLibraryItems( ctx: await ctx(), parentId: input.parentId, includeItemTypes: input.includeItemTypes, genreIds: input.genreIds, searchTerm: input.searchTerm, anyProviderIdEquals: input.anyProviderIdEquals, fields: input.fields, enableImageTypes: input.enableImageTypes, groupProgramsBySeries: input.groupProgramsBySeries, imageTypeLimit: input.imageTypeLimit, startIndex: input.startIndex, limit: input.limit, recursive: input.recursive, sortBy: input.sortBy, sortOrder: input.sortOrder, ); return LibraryItemsRes( parentId: input.parentId ?? '__global__', items: items, ); } } class GetHomeBannerUseCase extends AuthedUseCase { GetHomeBannerUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Future> execute({int fetchLimit = 24, int take = 8}) async { final items = await embyGateway.getLibraryItems( ctx: await ctx(), includeItemTypes: 'Movie,Series', recursive: true, sortBy: 'DateCreated', sortOrder: SortOrder.descending, fields: 'BasicSyncInfo,CommunityRating,ProductionYear,Overview,Genres,ProviderIds,UserData', enableImageTypes: 'Backdrop,Logo,Primary,Thumb', imageTypeLimit: 2, limit: fetchLimit, ); return items .where((it) => (it.BackdropImageTags ?? const []).isNotEmpty) .take(take) .toList(growable: false); } } class GlobalKeywordSearchUseCase extends AuthedUseCase { GlobalKeywordSearchUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); static const _defaultTypes = 'Movie,Series'; Stream executeStream(GlobalSearchReq input) { final controller = StreamController(); () async { final keyword = input.keyword.trim(); if (keyword.isEmpty) { controller.close(); return; } try { var sessions = await sessionRepository.listAll(); if (input.serverId != null) { sessions = sessions .where((s) => s.serverId == input.serverId) .toList(); } final allServers = await serverRepository.list(); final serverMap = {for (final s in allServers) s.id: s}; final types = input.includeItemTypes ?? _defaultTypes; final futures = sessions.map((session) async { final result = await _searchOne( session: session, server: serverMap[session.serverId], keyword: keyword, types: types, limit: input.limitPerServer ?? 30, ); if (result != null) controller.add(result); }); await Future.wait(futures); } catch (error, stackTrace) { controller.addError(error, stackTrace); } finally { controller.close(); } }(); return controller.stream; } Future _searchOne({ required SessionData session, required EmbyServer? server, required String keyword, required String types, required int limit, }) async { if (server == null) return null; final c = AuthedRequestContext( baseUrl: server.baseUrl, token: session.token, userId: session.userId, ); try { final items = await embyGateway.getLibraryItems( ctx: c, searchTerm: keyword, recursive: true, includeItemTypes: types, limit: limit, sortBy: 'SortName', sortOrder: SortOrder.ascending, ); if (items.isEmpty) return null; return GlobalSearchServerResult( serverId: server.id, serverName: server.name, serverUrl: server.baseUrl, token: session.token, userId: session.userId, items: items, ); } catch (e) { AppLogger.warn(_ucTag, 'GlobalKeywordSearch on ${server.name} failed', e); return null; } } } class GetAggregatedResumeUseCase extends AuthedUseCase { GetAggregatedResumeUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Stream executeStream({int? limitPerServer}) { return _fanOutPerServer( sessionRepository: sessionRepository, serverRepository: serverRepository, logTag: 'AggregatedResume', fetchItems: (context) async { final res = await embyGateway.getResumeItems( context, limit: limitPerServer ?? 20, ); return res.Items; }, ); } } class GetAggregatedFavoritesUseCase extends AuthedUseCase { static const _defaultTypes = 'Movie,Series,Episode'; GetAggregatedFavoritesUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Stream executeStream({ String includeItemTypes = _defaultTypes, int? limitPerServer, }) { final types = includeItemTypes.trim().isEmpty ? _defaultTypes : includeItemTypes; return _fanOutPerServer( sessionRepository: sessionRepository, serverRepository: serverRepository, logTag: 'AggregatedFavorites', fetchItems: (context) async { final items = await embyGateway.getFavoriteItems( ctx: context, includeItemTypes: types, sortBy: 'SortName', sortOrder: SortOrder.ascending, ); return limitPerServer == null ? items : items.take(limitPerServer).toList(growable: false); }, ); } } class GetAggregatedRecentUseCase extends AuthedUseCase { static const _defaultTypes = 'Movie,Series'; GetAggregatedRecentUseCase({ required super.sessionRepository, required super.serverRepository, required super.embyGateway, }); Stream executeStream({ String includeItemTypes = _defaultTypes, int? limitPerServer, }) { return _fanOutPerServer( sessionRepository: sessionRepository, serverRepository: serverRepository, logTag: 'AggregatedRecent', fetchItems: (context) async { return embyGateway.getLibraryItems( ctx: context, includeItemTypes: includeItemTypes, fields: 'BasicSyncInfo,CollectionType,PrimaryImageAspectRatio,UserData,' 'CommunityRating,ProviderIds,ProductionYear,ChildCount,Container,' 'CanDelete,DateCreated', sortBy: 'DateCreated', sortOrder: SortOrder.descending, recursive: true, limit: limitPerServer ?? 20, ); }, ); } }