From 8d5c0d5b25596bcfa14d11bcbdce2063b7b9643c Mon Sep 17 00:00:00 2001 From: Owais Date: Fri, 8 May 2026 21:15:24 -0500 Subject: [PATCH] refactor: streaming events for follow audit (#37) * feat: streamed follow audit emission * refactor: remove unused fetch method * feat: emit more granular status events * refactor: preserve state.visibleStatuses * empty ready/complete states expose Scan button --- lib/core/l10n/app_localizations.dart | 12 + lib/core/l10n/app_localizations_en.dart | 9 + lib/core/l10n/intl_en.arb | 14 + .../profile/cubit/follow_audit_cubit.dart | 156 +++++++--- .../profile/data/follow_audit_repository.dart | 111 +++++-- .../presentation/follow_audit_screen.dart | 144 +++++++-- .../cubit/follow_audit_cubit_test.dart | 274 ++++++++++++++---- .../data/follow_audit_repository_test.dart | 86 ++++-- .../follow_audit_integration_test.dart | 14 +- .../follow_audit_screen_test.dart | 44 ++- 10 files changed, 693 insertions(+), 171 deletions(-) diff --git a/lib/core/l10n/app_localizations.dart b/lib/core/l10n/app_localizations.dart index c8adcc3..00c43a6 100644 --- a/lib/core/l10n/app_localizations.dart +++ b/lib/core/l10n/app_localizations.dart @@ -2635,6 +2635,12 @@ abstract class AppLocalizations { /// **'Fetching follows: {progress}/{total}'** String formatFetchingFollowsProgress(int progress, int total); + /// Follow audit loading message while fetching the account follow count + /// + /// In en, this message translates to: + /// **'Getting follow count...'** + String get messageGettingFollowCount; + /// Snackbar message after following starter pack members /// /// In en, this message translates to: @@ -2647,6 +2653,12 @@ abstract class AppLocalizations { /// **'{count, plural, =1{1 follow scanned for problematic accounts} other{{count} follows scanned for problematic accounts}}'** String formatFollowsScanned(int count); + /// Follow audit intro message when the follow count is known + /// + /// In en, this message translates to: + /// **'Scan your {count, plural, =1{1 follow} other{{count} follows}} for deleted, suspended, blocked, and hidden accounts.'** + String formatFollowAuditPromptWithCount(int count); + /// Tooltip to hide a follow audit status /// /// In en, this message translates to: diff --git a/lib/core/l10n/app_localizations_en.dart b/lib/core/l10n/app_localizations_en.dart index 7ee47ef..7c0208a 100644 --- a/lib/core/l10n/app_localizations_en.dart +++ b/lib/core/l10n/app_localizations_en.dart @@ -1409,6 +1409,9 @@ class AppLocalizationsEn extends AppLocalizations { return 'Fetching follows: $progress/$total'; } + @override + String get messageGettingFollowCount => 'Getting follow count...'; + @override String formatFollowedMemberCount(int count) { String _temp0 = intl.Intl.pluralLogic(count, locale: localeName, other: '$count members', one: '1 member'); @@ -1426,6 +1429,12 @@ class AppLocalizationsEn extends AppLocalizations { return '$_temp0'; } + @override + String formatFollowAuditPromptWithCount(int count) { + String _temp0 = intl.Intl.pluralLogic(count, locale: localeName, other: '$count follows', one: '1 follow'); + return 'Scan your $_temp0 for deleted, suspended, blocked, and hidden accounts.'; + } + @override String formatHideStatus(String status) { return 'Hide $status'; diff --git a/lib/core/l10n/intl_en.arb b/lib/core/l10n/intl_en.arb index 9adbd67..ff3b3cb 100644 --- a/lib/core/l10n/intl_en.arb +++ b/lib/core/l10n/intl_en.arb @@ -1969,6 +1969,10 @@ } } }, + "messageGettingFollowCount": "Getting follow count...", + "@messageGettingFollowCount": { + "description": "Follow audit loading message while fetching the account follow count" + }, "formatFollowedMemberCount": "Followed {count, plural, =1{1 member} other{{count} members}}", "@formatFollowedMemberCount": { "description": "Snackbar message after following starter pack members", @@ -1989,6 +1993,16 @@ } } }, + "formatFollowAuditPromptWithCount": "Scan your {count, plural, =1{1 follow} other{{count} follows}} for deleted, suspended, blocked, and hidden accounts.", + "@formatFollowAuditPromptWithCount": { + "description": "Follow audit intro message when the follow count is known", + "placeholders": { + "count": { + "type": "int", + "example": "788" + } + } + }, "formatHideStatus": "Hide {status}", "@formatHideStatus": { "description": "Tooltip to hide a follow audit status", diff --git a/lib/features/profile/cubit/follow_audit_cubit.dart b/lib/features/profile/cubit/follow_audit_cubit.dart index efdf0c1..4ec5964 100644 --- a/lib/features/profile/cubit/follow_audit_cubit.dart +++ b/lib/features/profile/cubit/follow_audit_cubit.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:equatable/equatable.dart'; import 'package:flutter_bloc/flutter_bloc.dart'; import 'package:lazurite/core/logging/app_logger.dart'; @@ -73,49 +75,123 @@ class FollowAuditCubit extends Cubit { final FollowAuditRepository _repository; final String _ownDid; + StreamSubscription? _scanSubscription; + Completer? _scanCompleter; + int _auditGeneration = 0; + + @override + Future close() { + _auditGeneration++; + unawaited(_scanSubscription?.cancel()); + _scanSubscription = null; + if (_scanCompleter?.isCompleted == false) { + _scanCompleter?.complete(); + } + _scanCompleter = null; + return super.close(); + } - /// Fetches all follows then classifies them, emitting progress states along the way. + /// Scans follows and appends classified results as each page is processed. Future audit() async { - emit(state.copyWith(status: FollowAuditStatus.fetching, progress: 0, clearError: true)); + unawaited(_scanSubscription?.cancel()); + if (_scanCompleter?.isCompleted == false) { + _scanCompleter?.complete(); + } + final generation = ++_auditGeneration; + final completer = Completer(); + _scanCompleter = completer; - List records; - try { - records = await _repository.fetchAllFollows( - _ownDid, - onProgress: (fetched) { - emit(state.copyWith(status: FollowAuditStatus.fetching, progress: fetched)); - }, - ); - } catch (error, stackTrace) { - log.e('FollowAuditCubit: fetch failed', error: error, stackTrace: stackTrace); - emit(state.copyWith(status: FollowAuditStatus.error, errorMessage: error.toString())); + emit( + state.copyWith( + status: FollowAuditStatus.fetching, + results: const [], + totalFollows: 0, + progress: 0, + failedProfiles: 0, + unfollowedCount: 0, + visibleStatuses: FollowStatus.values.toSet(), + clearError: true, + ), + ); + + _scanSubscription = _repository + .scanFollows(_ownDid) + .listen( + (batch) { + if (!_isActiveAudit(generation)) { + return; + } + emit( + state.copyWith( + status: batch.isComplete ? FollowAuditStatus.ready : FollowAuditStatus.classifying, + results: [...state.results, ...batch.results], + totalFollows: _displayTotalFor(batch), + progress: batch.classifiedCount, + failedProfiles: batch.failedCount, + visibleStatuses: state.visibleStatuses.isEmpty ? FollowStatus.values.toSet() : state.visibleStatuses, + ), + ); + }, + onError: (Object error, StackTrace stackTrace) { + log.e('FollowAuditCubit: scan failed', error: error, stackTrace: stackTrace); + if (_isActiveAudit(generation)) { + emit(state.copyWith(status: FollowAuditStatus.error, errorMessage: error.toString())); + } + if (!completer.isCompleted) { + completer.complete(); + } + if (identical(_scanCompleter, completer)) { + _scanCompleter = null; + } + }, + onDone: () { + if (_isActiveAudit(generation) && + (state.status == FollowAuditStatus.fetching || state.status == FollowAuditStatus.classifying)) { + emit(state.copyWith(status: FollowAuditStatus.ready)); + } + if (_isActiveAudit(generation)) { + _scanSubscription = null; + } + if (!completer.isCompleted) { + completer.complete(); + } + if (identical(_scanCompleter, completer)) { + _scanCompleter = null; + } + }, + cancelOnError: true, + ); + + await completer.future; + } + + Future cancelAudit() async { + final subscription = _scanSubscription; + if (subscription == null || + (state.status != FollowAuditStatus.fetching && state.status != FollowAuditStatus.classifying)) { return; } - emit(state.copyWith(status: FollowAuditStatus.classifying, totalFollows: records.length, progress: 0)); + _auditGeneration++; + unawaited(subscription.cancel()); + if (identical(_scanSubscription, subscription)) { + _scanSubscription = null; + } + if (_scanCompleter?.isCompleted == false) { + _scanCompleter?.complete(); + } + _scanCompleter = null; + emit(state.copyWith(status: FollowAuditStatus.ready)); + } - try { - final (:results, :failedCount) = await _repository.classifyFollows( - records, - _ownDid, - onProgress: (classified) { - emit(state.copyWith(status: FollowAuditStatus.classifying, progress: classified)); - }, - ); + bool _isActiveAudit(int generation) => !isClosed && generation == _auditGeneration; - emit( - state.copyWith( - status: FollowAuditStatus.ready, - results: results, - failedProfiles: failedCount, - progress: records.length, - visibleStatuses: FollowStatus.values.toSet(), - ), - ); - } catch (error, stackTrace) { - log.e('FollowAuditCubit: classify failed', error: error, stackTrace: stackTrace); - emit(state.copyWith(status: FollowAuditStatus.error, errorMessage: error.toString())); + int _displayTotalFor(FollowAuditBatch batch) { + final expectedTotal = batch.totalFollows; + if (expectedTotal == null) { + return batch.scannedCount; } + return expectedTotal < batch.scannedCount ? batch.scannedCount : expectedTotal; } /// Toggles the selection of the result at [index]. @@ -160,8 +236,18 @@ class FollowAuditCubit extends Cubit { final count = await _repository.batchUnfollow(selected, _ownDid); final selectedUris = selected.map((r) => r.record.uri).toSet(); final remaining = state.results.where((r) => !selectedUris.contains(r.record.uri)).toList(); + final updatedTotal = (state.totalFollows - count).clamp(0, state.totalFollows); + final updatedProgress = (state.progress - count).clamp(0, state.progress); - emit(state.copyWith(status: FollowAuditStatus.complete, results: remaining, unfollowedCount: count)); + emit( + state.copyWith( + status: FollowAuditStatus.complete, + results: remaining, + totalFollows: updatedTotal, + progress: updatedProgress, + unfollowedCount: count, + ), + ); } catch (error, stackTrace) { log.e('FollowAuditCubit: unfollow failed', error: error, stackTrace: stackTrace); emit(state.copyWith(status: FollowAuditStatus.error, errorMessage: error.toString())); diff --git a/lib/features/profile/data/follow_audit_repository.dart b/lib/features/profile/data/follow_audit_repository.dart index f07f862..c112d85 100644 --- a/lib/features/profile/data/follow_audit_repository.dart +++ b/lib/features/profile/data/follow_audit_repository.dart @@ -17,6 +17,31 @@ class FollowRecord { final String subjectDid; } +class FollowRecordPage { + const FollowRecordPage({required this.records, required this.cursor}); + + final List records; + final String? cursor; +} + +class FollowAuditBatch { + const FollowAuditBatch({ + required this.totalFollows, + required this.scannedCount, + required this.classifiedCount, + required this.results, + required this.failedCount, + required this.isComplete, + }); + + final int? totalFollows; + final int scannedCount; + final int classifiedCount; + final List results; + final int failedCount; + final bool isComplete; +} + class ClassifiedFollow extends Equatable { const ClassifiedFollow({ required this.record, @@ -67,31 +92,79 @@ class FollowAuditRepository { final dynamic _bluesky; final AppViewRequestContext _appViewContext; - Future> fetchAllFollows(String did, {void Function(int fetched)? onProgress}) async { - _assertCurrentSessionRepoAccess(did: did, operation: 'fetchAllFollows'); + Future fetchFollowCount(String did) async { + _assertCurrentSessionRepoAccess(did: did, operation: 'fetchFollowCount'); + try { + final response = await _bluesky.actor.getProfile( + actor: did, + $headers: _appViewContext.appBskyHeadersForEndpoint('app.bsky.actor.getProfile'), + ); + final count = response.data.followsCount; + return count is int && count >= 0 ? count : null; + } catch (error, stackTrace) { + log.w('FollowAuditRepository: failed to fetch followsCount for $did', error: error, stackTrace: stackTrace); + return null; + } + } + + Future fetchFollowPage(String did, {String? cursor, int limit = 100}) async { + _assertCurrentSessionRepoAccess(did: did, operation: 'fetchFollowPage'); + final response = await _bluesky.atproto.repo.listRecords( + repo: did, + collection: 'app.bsky.graph.follow', + limit: limit.clamp(1, 100), + cursor: cursor, + ); + final records = []; + final rawRecords = response.data.records as List; + for (final raw in rawRecords) { + final uri = raw.uri.toString(); + final rkey = AtUri.parse(uri).rkey; + final subjectDid = raw.value['subject'] as String; + records.add(FollowRecord(uri: uri, rkey: rkey, subjectDid: subjectDid)); + } + + return FollowRecordPage(records: records, cursor: response.data.cursor as String?); + } + + Stream scanFollows(String did) async* { + _assertCurrentSessionRepoAccess(did: did, operation: 'scanFollows'); + final expectedTotalFollows = await fetchFollowCount(did); + var scannedCount = 0; + var classifiedCount = 0; + var failedCount = 0; String? cursor; do { - final response = await _bluesky.atproto.repo.listRecords( - repo: did, - collection: 'app.bsky.graph.follow', - limit: 100, - cursor: cursor, - ); - - final rawRecords = response.data.records as List; - for (final raw in rawRecords) { - final uri = raw.uri.toString(); - final rkey = AtUri.parse(uri).rkey; - final subjectDid = raw.value['subject'] as String; - records.add(FollowRecord(uri: uri, rkey: rkey, subjectDid: subjectDid)); + final page = await fetchFollowPage(did, cursor: cursor); + cursor = page.cursor; + scannedCount += page.records.length; + + if (page.records.isEmpty) { + yield FollowAuditBatch( + totalFollows: expectedTotalFollows, + scannedCount: scannedCount, + classifiedCount: classifiedCount, + results: const [], + failedCount: failedCount, + isComplete: cursor == null, + ); + continue; } - cursor = response.data.cursor as String?; - onProgress?.call(records.length); - } while (cursor != null); - return records; + final classified = await classifyFollows(page.records, did); + classifiedCount += page.records.length; + failedCount += classified.failedCount; + yield FollowAuditBatch( + totalFollows: expectedTotalFollows, + scannedCount: scannedCount, + classifiedCount: classifiedCount, + results: classified.results, + failedCount: failedCount, + isComplete: cursor == null, + ); + } while (cursor != null); } Future<({List results, int failedCount})> classifyFollows( diff --git a/lib/features/profile/presentation/follow_audit_screen.dart b/lib/features/profile/presentation/follow_audit_screen.dart index 7e2450e..f6da3af 100644 --- a/lib/features/profile/presentation/follow_audit_screen.dart +++ b/lib/features/profile/presentation/follow_audit_screen.dart @@ -1,10 +1,10 @@ import 'dart:math' as math; -import 'package:lazurite/core/l10n/l10n.dart'; -import 'package:lazurite/core/theme/theme_extensions.dart'; import 'package:flutter/material.dart'; import 'package:flutter/services.dart'; import 'package:flutter_bloc/flutter_bloc.dart'; +import 'package:lazurite/core/l10n/l10n.dart'; +import 'package:lazurite/core/theme/theme_extensions.dart'; import 'package:lazurite/features/profile/cubit/follow_audit_cubit.dart'; import 'package:lazurite/features/profile/data/follow_audit_repository.dart'; import 'package:lazurite/shared/presentation/helpers/navigation_helpers.dart'; @@ -26,13 +26,18 @@ class FollowAuditScreen extends StatelessWidget { return Column( children: [ - _HeaderCard(totalFollows: state.totalFollows), + _HeaderCard(scannedFollows: state.progress, totalFollows: state.totalFollows, status: state.status), Padding( padding: const EdgeInsets.fromLTRB(16, 12, 16, 0), child: Row( children: [ Expanded( - child: _AuditActionButton(status: state.status, selectedCount: selectedCount, isBusy: isBusy), + child: _AuditActionButton( + status: state.status, + selectedCount: selectedCount, + resultCount: state.results.length, + isBusy: isBusy, + ), ), ], ), @@ -130,12 +135,15 @@ class FollowAuditScreen extends StatelessWidget { } class _HeaderCard extends StatelessWidget { - const _HeaderCard({required this.totalFollows}); + const _HeaderCard({required this.scannedFollows, required this.totalFollows, required this.status}); + final int scannedFollows; final int totalFollows; + final FollowAuditStatus status; @override Widget build(BuildContext context) { + final subtitle = _subtitle(context); return Container( width: double.infinity, margin: const EdgeInsets.fromLTRB(16, 12, 16, 0), @@ -152,28 +160,61 @@ class _HeaderCard extends StatelessWidget { style: context.textTheme.labelLarge?.copyWith(fontWeight: FontWeight.w700, letterSpacing: 1.1), ), const SizedBox(height: 6), - Text( - totalFollows > 0 ? context.l10n.formatFollowsScanned(totalFollows) : context.l10n.messageFollowAuditIntro, - style: context.textTheme.bodyMedium, - ), + Text(subtitle, style: context.textTheme.bodyMedium), ], ), ); } + + String _subtitle(BuildContext context) { + if (status == FollowAuditStatus.fetching) { + return context.l10n.messageGettingFollowCount; + } + if (status == FollowAuditStatus.complete && totalFollows > 0) { + return context.l10n.formatFollowAuditPromptWithCount(totalFollows); + } + if (totalFollows > 0 && scannedFollows <= 0) { + return context.l10n.formatFollowAuditPromptWithCount(totalFollows); + } + if (totalFollows <= 0 && scannedFollows <= 0) { + return context.l10n.messageFollowAuditIntro; + } + if (status == FollowAuditStatus.classifying && totalFollows > 0) { + return context.l10n.formatClassifyingProgress(scannedFollows, totalFollows); + } + return context.l10n.formatFollowsScanned(scannedFollows); + } } class _AuditActionButton extends StatelessWidget { - const _AuditActionButton({required this.status, required this.selectedCount, required this.isBusy}); + const _AuditActionButton({ + required this.status, + required this.selectedCount, + required this.resultCount, + required this.isBusy, + }); final FollowAuditStatus status; final int selectedCount; + final int resultCount; final bool isBusy; @override Widget build(BuildContext context) { + if (status == FollowAuditStatus.fetching || status == FollowAuditStatus.classifying) { + return OutlinedButton.icon( + key: const Key('follow_audit_cancel_button'), + onPressed: () => context.read().cancelAudit(), + icon: const Icon(Icons.stop_circle_outlined), + label: Text(context.l10n.buttonCancel), + ); + } + if (status == FollowAuditStatus.initial || - status == FollowAuditStatus.fetching || - status == FollowAuditStatus.classifying) { + (resultCount == 0 && + selectedCount == 0 && + status != FollowAuditStatus.unfollowing && + status != FollowAuditStatus.error)) { return FilledButton.icon( key: const Key('follow_audit_scan_button'), onPressed: isBusy ? null : () => context.read().audit(), @@ -205,9 +246,10 @@ class _ProgressSection extends StatelessWidget { } final shownTotal = totalFollows > 0 ? totalFollows : math.max(progress, 1); - final value = (progress / shownTotal).clamp(0.0, 1.0); - final label = status == FollowAuditStatus.fetching - ? context.l10n.formatFetchingFollowsProgress(progress, shownTotal) + final isFetchingCount = status == FollowAuditStatus.fetching; + final value = isFetchingCount ? null : (progress / shownTotal).clamp(0.0, 1.0); + final label = isFetchingCount + ? context.l10n.messageGettingFollowCount : context.l10n.formatClassifyingProgress(progress, shownTotal); return Padding( @@ -439,10 +481,27 @@ class _ResultsPanelState extends State<_ResultsPanel> { return Center(child: Text(context.l10n.messageFollowAuditStartPrompt)); } + if (widget.state.status == FollowAuditStatus.fetching) { + return _CenteredAuditSpinner(message: context.l10n.messageGettingFollowCount); + } + if (widget.state.results.isEmpty && (widget.state.status == FollowAuditStatus.ready || widget.state.status == FollowAuditStatus.complete)) { return Center( - child: Text(context.l10n.messageNoProblematicFollows, key: const Key('follow_audit_empty_message')), + child: Text( + widget.state.status == FollowAuditStatus.complete && widget.state.totalFollows > 0 + ? context.l10n.formatFollowAuditPromptWithCount(widget.state.totalFollows) + : context.l10n.messageNoProblematicFollows, + key: const Key('follow_audit_empty_message'), + textAlign: TextAlign.center, + ), + ); + } + + if (widget.state.results.isEmpty && widget.state.status == FollowAuditStatus.classifying) { + return _CenteredAuditSpinner( + key: const Key('follow_audit_streaming_empty_message'), + message: context.l10n.formatClassifyingProgress(widget.state.progress, math.max(widget.state.totalFollows, 1)), ); } @@ -452,9 +511,12 @@ class _ResultsPanelState extends State<_ResultsPanel> { return ListView.separated( padding: const EdgeInsets.fromLTRB(16, 0, 16, 16), - itemCount: widget.visibleEntries.length, + itemCount: widget.visibleEntries.length + (widget.state.status == FollowAuditStatus.classifying ? 1 : 0), separatorBuilder: (context, index) => const SizedBox(height: 8), itemBuilder: (context, index) { + if (index >= widget.visibleEntries.length) { + return _ScanningFooter(state: widget.state); + } final entry = widget.visibleEntries[index]; final rowKey = '${entry.item.record.subjectDid}:${entry.item.record.rkey}'; return StaggeredEntrance( @@ -468,6 +530,54 @@ class _ResultsPanelState extends State<_ResultsPanel> { } } +class _CenteredAuditSpinner extends StatelessWidget { + const _CenteredAuditSpinner({super.key, required this.message}); + + final String message; + + @override + Widget build(BuildContext context) { + return Center( + child: Column( + mainAxisSize: MainAxisSize.min, + children: [ + const SizedBox(width: 28, height: 28, child: CircularProgressIndicator(strokeWidth: 2)), + const SizedBox(height: 12), + Text(message, textAlign: TextAlign.center, style: context.textTheme.bodyMedium), + ], + ), + ); + } +} + +class _ScanningFooter extends StatelessWidget { + const _ScanningFooter({required this.state}); + + final FollowAuditState state; + + @override + Widget build(BuildContext context) { + return Padding( + padding: const EdgeInsets.symmetric(vertical: 12), + child: Row( + mainAxisAlignment: MainAxisAlignment.center, + children: [ + const SizedBox(width: 16, height: 16, child: CircularProgressIndicator(strokeWidth: 2)), + const SizedBox(width: 8), + Flexible( + child: Text( + context.l10n.formatClassifyingProgress(state.progress, math.max(state.totalFollows, 1)), + key: const Key('follow_audit_scanning_footer'), + textAlign: TextAlign.center, + style: context.textTheme.bodySmall?.copyWith(color: context.colorScheme.onSurfaceVariant), + ), + ), + ], + ), + ); + } +} + class _ResultRow extends StatelessWidget { const _ResultRow({required this.index, required this.item}); diff --git a/test/features/profile/cubit/follow_audit_cubit_test.dart b/test/features/profile/cubit/follow_audit_cubit_test.dart index 219229e..9e1fadd 100644 --- a/test/features/profile/cubit/follow_audit_cubit_test.dart +++ b/test/features/profile/cubit/follow_audit_cubit_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:bloc_test/bloc_test.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:lazurite/features/profile/cubit/follow_audit_cubit.dart'; @@ -5,45 +7,37 @@ import 'package:lazurite/features/profile/data/follow_audit_repository.dart'; class _FakeFollowAuditRepository implements FollowAuditRepository { _FakeFollowAuditRepository({ - List fetchResult = const [], List classifyResult = const [], + List? scanBatches, int classifyFailedCount = 0, int batchUnfollowResult = 0, - Exception? fetchError, Exception? classifyError, + Exception? scanError, Exception? unfollowError, - List? fetchProgressValues, List? classifyProgressValues, - }) : _fetchResult = fetchResult, - _classifyResult = classifyResult, + Completer? scanGate, + Future Function(int batchIndex)? afterBatch, + }) : _classifyResult = classifyResult, + _scanBatches = scanBatches, _classifyFailedCount = classifyFailedCount, _batchUnfollowResult = batchUnfollowResult, - _fetchError = fetchError, _classifyError = classifyError, + _scanError = scanError, _unfollowError = unfollowError, - _fetchProgressValues = fetchProgressValues, - _classifyProgressValues = classifyProgressValues; + _classifyProgressValues = classifyProgressValues, + _scanGate = scanGate, + _afterBatch = afterBatch; - final List _fetchResult; final List _classifyResult; + final List? _scanBatches; final int _classifyFailedCount; final int _batchUnfollowResult; - final Exception? _fetchError; final Exception? _classifyError; + final Exception? _scanError; final Exception? _unfollowError; - final List? _fetchProgressValues; final List? _classifyProgressValues; - - @override - Future> fetchAllFollows(String did, {void Function(int fetched)? onProgress}) async { - if (_fetchError != null) throw _fetchError; - if (_fetchProgressValues != null) { - for (final v in _fetchProgressValues) { - onProgress?.call(v); - } - } - return _fetchResult; - } + final Completer? _scanGate; + final Future Function(int batchIndex)? _afterBatch; @override Future<({List results, int failedCount})> classifyFollows( @@ -66,6 +60,34 @@ class _FakeFollowAuditRepository implements FollowAuditRepository { return _batchUnfollowResult; } + @override + Stream scanFollows(String did) async* { + if (_scanError != null) throw _scanError; + + final batches = _scanBatches; + if (batches != null) { + for (var i = 0; i < batches.length; i++) { + final batch = batches[i]; + yield batch; + await _afterBatch?.call(i); + await Future.delayed(Duration.zero); + } + await _scanGate?.future; + return; + } + + if (_classifyError != null) throw _classifyError; + yield FollowAuditBatch( + totalFollows: _classifyResult.length, + scannedCount: _classifyResult.length, + classifiedCount: _classifyResult.length, + results: _classifyResult, + failedCount: _classifyFailedCount, + isComplete: true, + ); + await _scanGate?.future; + } + @override dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); } @@ -125,17 +147,14 @@ void main() { group('FollowAuditCubit.audit', () { blocTest( - 'transitions initial → fetching → classifying → ready', - build: () => _cubit( - _FakeFollowAuditRepository( - fetchResult: [_record('did:plc:alice')], - classifyResult: [_classified('did:plc:alice', FollowStatus.blockedBy)], - ), - ), + 'transitions initial → fetching → ready', + build: () => + _cubit(_FakeFollowAuditRepository(classifyResult: [_classified('did:plc:alice', FollowStatus.blockedBy)])), act: (cubit) => cubit.audit(), expect: () => [ - isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), - isA().having((s) => s.status, 'status', FollowAuditStatus.classifying), + isA() + .having((s) => s.status, 'status', FollowAuditStatus.fetching) + .having((s) => s.results, 'results', isEmpty), isA() .having((s) => s.status, 'status', FollowAuditStatus.ready) .having((s) => s.results.length, 'results', 1) @@ -144,61 +163,172 @@ void main() { ); blocTest( - 'emits progress updates during fetch phase', + 'appends streamed classified batches', build: () => _cubit( _FakeFollowAuditRepository( - fetchResult: [_record('did:plc:a'), _record('did:plc:b'), _record('did:plc:c')], - fetchProgressValues: [1, 2, 3], - classifyResult: [], + scanBatches: [ + FollowAuditBatch( + totalFollows: 3, + scannedCount: 2, + classifiedCount: 2, + results: [_classified('did:plc:a', FollowStatus.deleted)], + failedCount: 0, + isComplete: false, + ), + FollowAuditBatch( + totalFollows: 3, + scannedCount: 3, + classifiedCount: 3, + results: [_classified('did:plc:b', FollowStatus.blockedBy)], + failedCount: 1, + isComplete: true, + ), + ], ), ), act: (cubit) => cubit.audit(), expect: () => [ isA() .having((s) => s.status, 'status', FollowAuditStatus.fetching) - .having((s) => s.progress, 'progress', 0), + .having((s) => s.results, 'results', isEmpty), isA() - .having((s) => s.status, 'status', FollowAuditStatus.fetching) - .having((s) => s.progress, 'progress', 1), - isA() - .having((s) => s.status, 'status', FollowAuditStatus.fetching) - .having((s) => s.progress, 'progress', 2), + .having((s) => s.status, 'status', FollowAuditStatus.classifying) + .having((s) => s.progress, 'progress', 2) + .having((s) => s.totalFollows, 'totalFollows', 3) + .having((s) => s.results.length, 'results', 1), isA() - .having((s) => s.status, 'status', FollowAuditStatus.fetching) - .having((s) => s.progress, 'progress', 3), - isA().having((s) => s.status, 'status', FollowAuditStatus.classifying), - isA().having((s) => s.status, 'status', FollowAuditStatus.ready), + .having((s) => s.status, 'status', FollowAuditStatus.ready) + .having((s) => s.progress, 'progress', 3) + .having((s) => s.totalFollows, 'totalFollows', 3) + .having((s) => s.results.length, 'results', 2) + .having((s) => s.failedProfiles, 'failedProfiles', 1), ], ); blocTest( - 'emits progress updates during classify phase', + 'can cancel streaming scan and keep partial results ready for unfollow', build: () => _cubit( _FakeFollowAuditRepository( - fetchResult: [_record('did:plc:a'), _record('did:plc:b')], - classifyProgressValues: [1, 2], - classifyResult: [], + scanBatches: [ + FollowAuditBatch( + totalFollows: 5, + scannedCount: 1, + classifiedCount: 1, + results: [_classified('did:plc:a', FollowStatus.deleted).copyWith(selected: true)], + failedCount: 0, + isComplete: false, + ), + ], + scanGate: Completer(), ), ), - act: (cubit) => cubit.audit(), + act: (cubit) async { + unawaited(cubit.audit()); + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + await cubit.cancelAudit(); + }, expect: () => [ - isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), isA() - .having((s) => s.status, 'status', FollowAuditStatus.classifying) - .having((s) => s.progress, 'progress', 0), + .having((s) => s.status, 'status', FollowAuditStatus.fetching) + .having((s) => s.results, 'results', isEmpty), isA() .having((s) => s.status, 'status', FollowAuditStatus.classifying) .having((s) => s.progress, 'progress', 1), + isA() + .having((s) => s.status, 'status', FollowAuditStatus.ready) + .having((s) => s.results.length, 'results', 1), + ], + ); + + blocTest( + 'uses expected follow count as progress denominator while scan is incomplete', + build: () => _cubit( + _FakeFollowAuditRepository( + scanBatches: [ + FollowAuditBatch( + totalFollows: 789, + scannedCount: 100, + classifiedCount: 100, + results: [_classified('did:plc:a', FollowStatus.deleted)], + failedCount: 0, + isComplete: false, + ), + ], + scanGate: Completer(), + ), + ), + act: (cubit) async { + unawaited(cubit.audit()); + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + await cubit.cancelAudit(); + }, + expect: () => [ + isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), isA() .having((s) => s.status, 'status', FollowAuditStatus.classifying) - .having((s) => s.progress, 'progress', 2), + .having((s) => s.progress, 'progress', 100) + .having((s) => s.totalFollows, 'totalFollows', 789), isA().having((s) => s.status, 'status', FollowAuditStatus.ready), ], ); + late Completer releaseSecondBatch; + blocTest( + 'preserves visibility filters across streamed batches', + build: () { + releaseSecondBatch = Completer(); + return _cubit( + _FakeFollowAuditRepository( + scanBatches: [ + FollowAuditBatch( + totalFollows: 2, + scannedCount: 1, + classifiedCount: 1, + results: [_classified('did:plc:a', FollowStatus.deleted)], + failedCount: 0, + isComplete: false, + ), + FollowAuditBatch( + totalFollows: 2, + scannedCount: 2, + classifiedCount: 2, + results: [_classified('did:plc:b', FollowStatus.blockedBy)], + failedCount: 0, + isComplete: true, + ), + ], + afterBatch: (index) => index == 0 ? releaseSecondBatch.future : Future.value(), + ), + ); + }, + act: (cubit) async { + unawaited(cubit.audit()); + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + cubit.toggleVisibility(FollowStatus.blockedBy); + releaseSecondBatch.complete(); + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + }, + expect: () => [ + isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), + isA().having((s) => s.status, 'status', FollowAuditStatus.classifying), + isA().having( + (s) => s.visibleStatuses, + 'visibleStatuses', + isNot(contains(FollowStatus.blockedBy)), + ), + isA() + .having((s) => s.status, 'status', FollowAuditStatus.ready) + .having((s) => s.visibleStatuses, 'visibleStatuses', isNot(contains(FollowStatus.blockedBy))), + ], + ); + blocTest( - 'transitions to error when fetch fails', - build: () => _cubit(_FakeFollowAuditRepository(fetchError: Exception('network error'))), + 'transitions to error when scan fails', + build: () => _cubit(_FakeFollowAuditRepository(scanError: Exception('network error'))), act: (cubit) => cubit.audit(), expect: () => [ isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), @@ -211,12 +341,22 @@ void main() { blocTest( 'transitions to ready with empty results when no problematic follows found', build: () => _cubit( - _FakeFollowAuditRepository(fetchResult: [_record('did:plc:alice'), _record('did:plc:bob')], classifyResult: []), + _FakeFollowAuditRepository( + scanBatches: const [ + FollowAuditBatch( + totalFollows: 2, + scannedCount: 2, + classifiedCount: 2, + results: [], + failedCount: 0, + isComplete: true, + ), + ], + ), ), act: (cubit) => cubit.audit(), expect: () => [ isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), - isA().having((s) => s.status, 'status', FollowAuditStatus.classifying), isA() .having((s) => s.status, 'status', FollowAuditStatus.ready) .having((s) => s.results, 'results', isEmpty) @@ -227,12 +367,22 @@ void main() { blocTest( 'records failedProfiles from classify', build: () => _cubit( - _FakeFollowAuditRepository(fetchResult: [_record('did:plc:alice')], classifyResult: [], classifyFailedCount: 3), + _FakeFollowAuditRepository( + scanBatches: const [ + FollowAuditBatch( + totalFollows: 1, + scannedCount: 1, + classifiedCount: 1, + results: [], + failedCount: 3, + isComplete: true, + ), + ], + ), ), act: (cubit) => cubit.audit(), expect: () => [ isA().having((s) => s.status, 'status', FollowAuditStatus.fetching), - isA().having((s) => s.status, 'status', FollowAuditStatus.classifying), isA() .having((s) => s.status, 'status', FollowAuditStatus.ready) .having((s) => s.failedProfiles, 'failedProfiles', 3), @@ -342,6 +492,8 @@ void main() { }, seed: () => FollowAuditState( status: FollowAuditStatus.ready, + totalFollows: 10, + progress: 10, results: [ _classified('did:plc:a', FollowStatus.deleted).copyWith(selected: true), _classified('did:plc:b', FollowStatus.blockedBy).copyWith(selected: true), @@ -354,6 +506,8 @@ void main() { isA() .having((s) => s.status, 'status', FollowAuditStatus.complete) .having((s) => s.unfollowedCount, 'unfollowedCount', 2) + .having((s) => s.totalFollows, 'totalFollows', 8) + .having((s) => s.progress, 'progress', 8) .having((s) => s.results.length, 'results length', 1), ], ); diff --git a/test/features/profile/data/follow_audit_repository_test.dart b/test/features/profile/data/follow_audit_repository_test.dart index 4fcbd13..d7f530e 100644 --- a/test/features/profile/data/follow_audit_repository_test.dart +++ b/test/features/profile/data/follow_audit_repository_test.dart @@ -96,6 +96,7 @@ class _FakeActorService { _FakeActorService({ this.batchProfiles = const {}, this.singleProfiles = const {}, + this.followCounts = const {}, Map? singleErrors, this.batchFailCount = 0, }) : singleErrors = singleErrors ?? {}; @@ -106,6 +107,9 @@ class _FakeActorService { /// DID → ProfileView for per-DID fallback (getProfile). final Map singleProfiles; + /// DID → followsCount for profile count lookup. + final Map followCounts; + /// DID → exception for per-DID fallback. final Map singleErrors; @@ -126,13 +130,17 @@ class _FakeActorService { return _FakeResponse(_FakeProfilesOutput(profiles: matched)); } - Future<_FakeResponse> getProfile({ + Future<_FakeResponse> getProfile({ required String actor, String? $service, Map? $headers, }) async { final error = singleErrors[actor]; if (error != null) throw error; + final followCount = followCounts[actor]; + if (followCount != null) { + return _FakeResponse(ProfileViewDetailed(did: actor, handle: 'owner.bsky.social', followsCount: followCount)); + } final profile = singleProfiles[actor]; if (profile == null) throw Exception('HTTP 404 Profile not found'); return _FakeResponse(profile); @@ -155,6 +163,7 @@ _FakeBluesky _bluesky({ List> pages = const [], Map batchProfiles = const {}, Map singleProfiles = const {}, + Map followCounts = const {}, Map? singleErrors, int batchFailCount = 0, void Function(String repo, List writes)? applyWritesCallback, @@ -166,6 +175,7 @@ _FakeBluesky _bluesky({ actor: _FakeActorService( batchProfiles: batchProfiles, singleProfiles: singleProfiles, + followCounts: followCounts, singleErrors: singleErrors, batchFailCount: batchFailCount, ), @@ -263,58 +273,70 @@ void main() { }); }); - group('FollowAuditRepository.fetchAllFollows', () { - test('returns empty list when no follows', () async { - final client = _bluesky(pages: []); + group('FollowAuditRepository.fetchFollowPage', () { + test('returns one page of follow records and cursor', () async { + final client = _bluesky( + pages: [ + [(_uri('did:plc:alice', 'rkey123'), 'did:plc:alice')], + [(_uri('did:plc:bob', 'rkey456'), 'did:plc:bob')], + ], + ); final repo = _repo(client); - final result = await repo.fetchAllFollows(_ownerDid); + final page = await repo.fetchFollowPage(_ownerDid); - expect(result, isEmpty); + expect(page.records.length, 1); + expect(page.records.first.rkey, 'rkey123'); + expect(page.cursor, 'page1'); }); + }); - test('returns all records from a single page', () async { - final client = _bluesky( - pages: [ - [(_uri('did:plc:alice'), 'did:plc:alice'), (_uri('did:plc:bob'), 'did:plc:bob')], - ], - ); + group('FollowAuditRepository.fetchFollowCount', () { + test('returns followsCount from own profile', () async { + final client = _bluesky(followCounts: {_ownerDid: 789}); final repo = _repo(client); - final result = await repo.fetchAllFollows(_ownerDid); + final count = await repo.fetchFollowCount(_ownerDid); - expect(result.length, 2); - expect(result[0].subjectDid, 'did:plc:alice'); - expect(result[1].subjectDid, 'did:plc:bob'); + expect(count, 789); }); - test('extracts rkey from AT URI correctly', () async { - final client = _bluesky( - pages: [ - [(_uri('did:plc:alice', 'rkey123'), 'did:plc:alice')], - ], - ); + test('returns null when followsCount cannot be loaded', () async { + final client = _bluesky(); final repo = _repo(client); - final result = await repo.fetchAllFollows(_ownerDid); + final count = await repo.fetchFollowCount(_ownerDid); - expect(result.first.rkey, 'rkey123'); + expect(count, isNull); }); + }); - test('paginates across multiple pages until cursor is null', () async { + group('FollowAuditRepository.scanFollows', () { + test('streams classified pages as they are processed', () async { + final aliceProfile = _profile('did:plc:alice', 'alice.bsky.social', blockedBy: true); + final bobProfile = _profile('did:plc:bob', 'bob.bsky.social', blocking: true); final client = _bluesky( pages: [ - [(_uri('did:plc:a1'), 'did:plc:a1')], - [(_uri('did:plc:a2'), 'did:plc:a2')], - [(_uri('did:plc:a3'), 'did:plc:a3')], + [(_uri('did:plc:alice'), 'did:plc:alice')], + [(_uri('did:plc:bob'), 'did:plc:bob')], ], + followCounts: {_ownerDid: 789}, + batchProfiles: {'did:plc:alice': aliceProfile, 'did:plc:bob': bobProfile}, ); final repo = _repo(client); - final result = await repo.fetchAllFollows(_ownerDid); - - expect(result.length, 3); - expect(result.map((r) => r.subjectDid), containsAll(['did:plc:a1', 'did:plc:a2', 'did:plc:a3'])); + final batches = await repo.scanFollows(_ownerDid).toList(); + + expect(batches.length, 2); + expect(batches.first.totalFollows, 789); + expect(batches.first.scannedCount, 1); + expect(batches.first.classifiedCount, 1); + expect(batches.first.results.single.status, FollowStatus.blockedBy); + expect(batches.first.isComplete, isFalse); + expect(batches.last.scannedCount, 2); + expect(batches.last.classifiedCount, 2); + expect(batches.last.results.single.status, FollowStatus.blocking); + expect(batches.last.isComplete, isTrue); }); }); diff --git a/test/features/profile/presentation/follow_audit_integration_test.dart b/test/features/profile/presentation/follow_audit_integration_test.dart index d3da757..f112893 100644 --- a/test/features/profile/presentation/follow_audit_integration_test.dart +++ b/test/features/profile/presentation/follow_audit_integration_test.dart @@ -31,11 +31,15 @@ class _ScriptedFollowAuditRepository implements FollowAuditRepository { } @override - Future> fetchAllFollows(String did, {void Function(int fetched)? onProgress}) async { - for (var i = 1; i <= records.length; i++) { - onProgress?.call(i); - } - return records; + Stream scanFollows(String did) async* { + yield FollowAuditBatch( + totalFollows: records.length, + scannedCount: records.length, + classifiedCount: records.length, + results: classified, + failedCount: 0, + isComplete: true, + ); } @override diff --git a/test/features/profile/presentation/follow_audit_screen_test.dart b/test/features/profile/presentation/follow_audit_screen_test.dart index 4291731..e19ad04 100644 --- a/test/features/profile/presentation/follow_audit_screen_test.dart +++ b/test/features/profile/presentation/follow_audit_screen_test.dart @@ -61,6 +61,7 @@ void main() { cubit = MockFollowAuditCubit(); when(() => cubit.audit()).thenAnswer((_) async {}); + when(() => cubit.cancelAudit()).thenAnswer((_) async {}); when(() => cubit.confirmUnfollow()).thenAnswer((_) async {}); when(() => cubit.toggleSelection(any())).thenReturn(null); when(() => cubit.selectAllByStatus(FollowStatus.deleted)).thenReturn(null); @@ -83,15 +84,37 @@ void main() { expect(find.text('Scan'), findsOneWidget); }); - testWidgets('fetching state shows progress bar with count text', (tester) async { - const state = FollowAuditState(status: FollowAuditStatus.fetching, progress: 3, totalFollows: 7); + testWidgets('fetching state shows getting follow count progress and cancel button', (tester) async { + const state = FollowAuditState(status: FollowAuditStatus.fetching); when(() => cubit.state).thenReturn(state); whenListen(cubit, const Stream.empty(), initialState: state); await tester.pumpWidget(_buildSubject(cubit)); expect(find.byKey(const Key('follow_audit_progress')), findsOneWidget); - expect(find.text('Fetching follows: 3/7'), findsOneWidget); + expect(find.text('Getting follow count...'), findsWidgets); + expect(find.byKey(const Key('follow_audit_cancel_button')), findsOneWidget); + }); + + testWidgets('classifying state shows cancel button and streams existing results', (tester) async { + final state = FollowAuditState( + status: FollowAuditStatus.classifying, + progress: 1, + totalFollows: 2, + results: [_classified(did: 'did:plc:alice', rkey: 'alice', status: FollowStatus.deleted, statusLabel: 'Deleted')], + visibleStatuses: FollowStatus.values.toSet(), + ); + when(() => cubit.state).thenReturn(state); + whenListen(cubit, const Stream.empty(), initialState: state); + + await tester.pumpWidget(_buildSubject(cubit)); + + expect(find.byKey(const Key('follow_audit_cancel_button')), findsOneWidget); + expect(find.text('alice.bsky.social'), findsOneWidget); + expect(find.byKey(const Key('follow_audit_scanning_footer')), findsOneWidget); + + await tester.tap(find.byKey(const Key('follow_audit_cancel_button'))); + verify(() => cubit.cancelAudit()).called(1); }); testWidgets('ready state renders results list with status badges', (tester) async { @@ -268,6 +291,21 @@ void main() { expect(find.text('No problematic follows found'), findsOneWidget); }); + testWidgets('complete state with no results shows updated follow count prompt', (tester) async { + const state = FollowAuditState(status: FollowAuditStatus.complete, totalFollows: 788, progress: 788); + when(() => cubit.state).thenReturn(state); + whenListen(cubit, const Stream.empty(), initialState: state); + + await tester.pumpWidget(_buildSubject(cubit)); + + expect(find.byKey(const Key('follow_audit_empty_message')), findsOneWidget); + expect(find.text('Scan your 788 follows for deleted, suspended, blocked, and hidden accounts.'), findsWidgets); + expect(find.byKey(const Key('follow_audit_scan_button')), findsOneWidget); + + await tester.tap(find.byKey(const Key('follow_audit_scan_button'))); + verify(() => cubit.audit()).called(1); + }); + testWidgets('tapping a handle navigates to profile screen', (tester) async { final results = [ _classified(did: 'did:plc:alice', rkey: 'alice', status: FollowStatus.deleted, statusLabel: 'Deleted'), -- 2.51.2