feat(llc)!: adopt the stream_core error layer and upload task API

docs/code_snippets/03_03_file_uploads.dart

@@ -20,17 +20,23 @@ Future<void> howToUploadAFileOrImageStep1() async {
     custom: {'width': 600, 'height': 400},
   );
 
-  // Upload the attachment
-  final result = await attachmentUploader.upload(
-    streamAttachment,
-    // Optionally track upload progress
-    onProgress: (progress) {
-      // Handle progress updates
-    },
-  );
-
-  // Map the result to an Attachment model to send with an activity
-  final uploadedAttachment = result.getOrThrow();
+  // Start the upload. The task comes back straight away, already running, and
+  // `task.cancel()` calls it off.
+  final task = attachmentUploader.upload(streamAttachment);
+
+  // Optionally track upload progress. `fraction` is null while the file's
+  // length is unknown, which is when an indeterminate bar is the right thing
+  // to show.
+  task.state.listen((state) {
+    if (state case UploadInProgress(progress: UploadProgress(:final fraction?))) {
+      print('${(fraction * 100).round()}% sent');
+    }
+  });
+
+  // Map the result to an Attachment model to send with an activity. A failed
+  // upload carries its error, so `getOrThrow` opts into throwing it; `fold`
+  // handles it instead.
+  final uploadedAttachment = (await task.result).getOrThrow();
   final attachmentReadyToBeSent = Attachment(
     imageUrl: uploadedAttachment.remoteUrl,
     assetUrl: uploadedAttachment.remoteUrl,

melos.yaml

@@ -25,6 +25,7 @@ command:
       auto_route: ^11.0.0
       collection: ^1.18.0
       chewie: ^1.11.3
+      clock: ^1.1.2
       dio: ^5.9.0
       equatable: ^2.0.5
       flutter_state_notifier: ^1.0.0
@@ -53,7 +54,7 @@ command:
       stream_core: 
         git:
           url: https://github.com/GetStream/stream-core-flutter.git
-          ref: f83b5d4d706a79fc429de2d27aead4394b83c1fb
+          ref: 40b1180f6b5f9ca53a5ecb2c74152fb1c6b4a376
           path: packages/stream_core
       video_player: ^2.10.0
       uuid: ^4.5.1
@@ -63,6 +64,7 @@ command:
     dev_dependencies:
       auto_route_generator: ^10.0.0
       build_runner: ^2.4.15
+      fake_async: ^1.3.3
       flutter_launcher_icons: ^0.14.4
       freezed: ^3.0.0
       injectable_generator: ^3.0.0

packages/stream_feeds/CHANGELOG.md

@@ -7,6 +7,10 @@
 - `PollResponseData.votingVisibility` is now required, so anything constructing one directly must supply it
 - `ActivityCommentList.state` returns `ActivityCommentListState` rather than `StateNotifier<ActivityCommentListState>`, matching the other state classes
 - Removed the call, recording, streaming and chat types that were never part of the Feeds API
+- Every failure the SDK reports for work it attempted now arrives as a `StreamException` subclass — `StreamApiException`, `StreamNetworkException`, `StreamAuthenticationException` or `StreamClientException` — replacing `ClientException` and `HttpClientException`, which are removed. `StreamApiError` remains, as the server's error payload and the type of `ConnectionErrorEvent.error`, but is no longer what the SDK throws or returns. `StreamFeedsException` aliases the base type, so one `on` clause catches all four
+- `connect` throws a `StateError` when a connection is already established or in progress, and a `StreamFeedsException` carrying the cause when it fails
+- `StreamAttachmentUploader.upload`, reached through `StreamFeedsClient.attachmentUploader`, returns an `AttachmentUploadTask` rather than a `Future<Result<UploadedAttachment>>`, and takes no `onProgress`: progress arrives on the task's `state`. `uploadBatch` returns an `AttachmentUploadBatch` rather than a `Stream<Result<UploadedAttachment>>`
+- `Feed.addActivity`, `Feed.addComment` and `Activity.addCommentsBatch` throw an `ArgumentError` when two attachments in one request share an id, rather than reporting it through the returned `Result`
 
 ### ✨ Features
 
@@ -29,15 +33,16 @@
 
 ### 🐛 Bug Fixes
 
+- Fixed a batch never running again after its first: an add that arrived once a batch had run joined that settled one instead of starting its own, so feed capabilities were fetched once per client and every feed discovered afterwards was answered with the first batch's result
 - Fixed `connect` failing when called straight after `disconnect`
 - Fixed a connection that could not authenticate hanging until it timed out, rather than failing with the reason
 - Fixed the `X-Stream-Client` header: the SDK identifier was sent twice, the version was hardcoded, and the OS was left out
 
 ### 🔄 Changed
 
+- Attachment uploads for a batch of requests now share one concurrency limit instead of one each, so `Activity.addCommentsBatch` no longer starts several uploads per comment at once; a failure also calls off the uploads still in flight rather than letting them finish work that is about to be discarded
 - `disconnect` now only closes the connection, leaving the client reusable with its existing subscriptions intact; releasing it is `dispose`
 - An expired token now recovers on its own: the connection comes back with one the `TokenProvider` issued afterwards, without the app doing anything
-- `connect` throws a `ClientException` when a connection is already established or in progress, and the one it throws on failure carries the underlying cause
 - Renamed the types below. The old names still compile, with a deprecation warning, and `dart fix --apply` migrates them:
 
 | Old name | New name |

packages/stream_feeds/lib/src/cdn/cdn_api.dart

@@ -45,5 +45,5 @@ abstract interface class CdnApi {
 
 class _ResultCallAdapter<T> extends CallAdapter<Future<T>, Future<Result<T>>> {
   @override
-  Future<Result<T>> adapt(Future<T> Function() call) => runSafely(call);
+  Future<Result<T>> adapt(Future<T> Function() call) => runApiSafely(call);
 }

packages/stream_feeds/lib/src/client/feeds_client_impl.dart

@@ -255,15 +255,18 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
 
   Future<void> _authenticateUser(
     WsRequestSender send,
-    StreamApiError? previousError,
+    StreamApiException? previousError,
   ) async {
-    if (previousError?.isTokenExpiredError ?? false) {
+    if (previousError?.isTokenExpired ?? false) {
       _tokenManager.expireToken();
 
       // A guest cannot refresh: another exchange answers with a different guest. The session ends
       // here, and the app starts another by building a new client.
       if (_tokenManager.usesStaticProvider) {
-        throw ClientException(message: 'The token was refused and the provider has no other to give');
+        throw StreamAuthenticationException(
+          message: 'The token was refused and the provider has no other to give',
+          cause: previousError,
+        );
       }
     }
 
@@ -297,11 +300,11 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
     }
 
     if (connectionState.value case Connecting() || Authenticating()) {
-      throw ClientException(message: 'Connection already in progress for ${user.id}');
+      throw StateError('Connection already in progress for ${user.id}');
     }
 
     if (connectionState.value case Connected()) {
-      throw ClientException(message: 'Connection already available for ${user.id}');
+      throw StateError('Connection already available for ${user.id}');
     }
 
     _logger.d(() => 'connect ${user.id} (${user.type.name}), webSocket: $connectWebSocket');
@@ -332,14 +335,16 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
   Future<void> _exchangeForGuestIdentity() async {
     final result = await _guestRepository.createGuest(user);
 
-    // Reported like every other connect failure, with the cause attached.
-    final response = result.getOrElse(
-      (error, stackTrace) => throw ClientException(
+    // Reported like every other connect failure, already classified by the API call seam.
+    final response = result.getOrElse((error, stackTrace) {
+      var exception = StreamException.tryFrom(error);
+      exception ??= StreamClientException(
         message: 'Failed to create a guest user',
-        error: error,
-        stackTrace: stackTrace,
-      ),
-    );
+        cause: error,
+      );
+
+      Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current);
+    });
 
     final tokenProvider = TokenProvider.static(response.token);
     _logger.d(() => 'guest created, server assigned ${response.user.id}');
@@ -370,7 +375,16 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
 
     if (state case Disconnected(:final source)) {
       _logger.w(() => 'connect ${user.id} failed: ${source.closeReason}', error: source.cause);
-      throw ClientException(message: source.closeReason, error: source.cause);
+
+      var exception = StreamException.tryFrom(source.cause);
+      exception ??= StreamNetworkException(message: source.closeReason, cause: source.cause);
+
+      final stackTrace = switch (source) {
+        ServerInitiated(:final stackTrace) || AuthenticationFailed(:final stackTrace) => stackTrace,
+        UserInitiated() || SystemInitiated() || UnHealthyConnection() || ConnectTimeout() => null,
+      };
+
+      Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current);
     }
 
     _logger.d(() => 'connected ${user.id}');

packages/stream_feeds/lib/src/feeds_client.dart

@@ -49,6 +49,12 @@ import 'state/user_list.dart';
 
 export 'client/moderation_client.dart';
 
+/// The root of the failures this SDK reports.
+///
+/// An alias of [StreamException], so code written against either name catches
+/// the same failures.
+typedef StreamFeedsException = StreamException;
+
 /// {@template stream_feeds_client}
 /// Stream Feeds client for building scalable newsfeeds and activity streams.
 ///
@@ -258,8 +264,9 @@ abstract interface class StreamFeedsClient {
 
   /// Establishes a connection to the Stream service.
   ///
-  /// Call this before anything else on the client. Throws a [ClientException] if the connection
-  /// fails, or if one is already established or in progress, and a [StateError] after [dispose].
+  /// Call this before anything else on the client. Throws a [StreamFeedsException] if the
+  /// connection fails, and a [StateError] when one is already established or in progress, or
+  /// after [dispose].
   ///
   /// Pass [connectWebSocket] as `false` if the client only needs to make requests. In that case:
   ///

packages/stream_feeds/lib/src/generated/api/api/default_api.dart

@@ -801,5 +801,5 @@ abstract interface class DefaultApi {
 
 class _ResultCallAdapter<T> extends CallAdapter<Future<T>, Future<Result<T>>> {
   @override
-  Future<Result<T>> adapt(Future<T> Function() call) => runSafely(call);
+  Future<Result<T>> adapt(Future<T> Function() call) => runApiSafely(call);
 }

packages/stream_feeds/lib/src/repository/capabilities_repository.dart

@@ -77,25 +77,15 @@ class CapabilitiesRepository {
 }
 
 extension on Result<Map<String, List<FeedOwnCapability>>> {
-  bool shouldRetry() {
-    switch (this) {
-      case api.Success():
-        return false;
-
-      case final api.Failure failure:
-        final error = failure.error;
-        if (error is! StreamDioException) {
-          return false;
-        }
-        final exception = error.exception;
-        if (exception is! HttpClientException) {
-          return false;
-        }
-        final statusCode = exception.statusCode;
-        if (statusCode == null) {
-          return false;
-        }
-        return statusCode < 100 || statusCode >= 500;
-    }
-  }
+  bool shouldRetry() => switch (this) {
+    api.Success() => false,
+    api.Failure(:final error) => switch (error) {
+      StreamNetworkException(isCancelled: true) => false,
+      StreamNetworkException() => true,
+      // A rate limit is not retried here: this waits a fixed moment, which is
+      // not the wait a rate limit asks for.
+      StreamApiException(:final statusCode) => statusCode < 100 || statusCode >= 500,
+      _ => false,
+    },
+  };
 }

packages/stream_feeds/lib/src/state/activity.dart

@@ -186,6 +186,16 @@ class Activity with Disposable {
 
   /// Adds a comment to this activity.
   ///
+  /// Attachments in [ActivityAddCommentRequest.attachmentUploads] are uploaded
+  /// first, and the comment is added once they are all in. To follow those
+  /// uploads or call them off, upload through
+  /// `StreamFeedsClient.attachmentUploader` instead and pass the results as
+  /// [ActivityAddCommentRequest.attachments].
+  ///
+  /// Throws an [ArgumentError] if two of those attachments share an id. Ids
+  /// default to a fresh UUID, so this only happens when one is given
+  /// explicitly, or the same attachment is listed twice.
+  ///
   /// Returns a [Result] containing the created [CommentData] or an error.
   Future<Result<CommentData>> addComment({
     required ActivityAddCommentRequest request,
@@ -203,6 +213,16 @@ class Activity with Disposable {
 
   /// Adds multiple comments to this activity in a batch.
   ///
+  /// Every request's attachments are uploaded as one batch, so a failure in one
+  /// can call off the uploads of the others. To follow those uploads or call
+  /// them off, upload through `StreamFeedsClient.attachmentUploader` instead
+  /// and pass the results as [ActivityAddCommentRequest.attachments].
+  ///
+  /// Throws an [ArgumentError] if two attachments share an id, across the whole
+  /// batch rather than within one request — one batch cannot tell them apart.
+  /// Ids default to a fresh UUID, so this only happens when one is given
+  /// explicitly, or the same attachment is listed twice.
+  ///
   /// Returns a [Result] containing a list of created [CommentData] or an error.
   Future<Result<List<CommentData>>> addCommentsBatch(
     List<ActivityAddCommentRequest> requests,

packages/stream_feeds/lib/src/state/feed.dart

@@ -186,6 +186,16 @@ class Feed with Disposable {
   ///
   /// The [request] contains the activity data to add.
   ///
+  /// Attachments in [FeedAddActivityRequest.attachmentUploads] are uploaded
+  /// first, and the activity is added once they are all in. To follow those
+  /// uploads or call them off, upload through
+  /// `StreamFeedsClient.attachmentUploader` instead and pass the results as
+  /// [FeedAddActivityRequest.attachments].
+  ///
+  /// Throws an [ArgumentError] if two of those attachments share an id. Ids
+  /// default to a fresh UUID, so this only happens when one is given
+  /// explicitly, or the same attachment is listed twice.
+  ///
   /// Returns a [Result] containing the added [ActivityData] if successful, or an error if the
   /// operation fails.
   Future<Result<ActivityData>> addActivity({

packages/stream_feeds/lib/src/state/feed_state.dart

@@ -1,5 +1,6 @@
 import 'dart:math';
 
+import 'package:clock/clock.dart';
 import 'package:collection/collection.dart';
 import 'package:freezed_annotation/freezed_annotation.dart';
 import 'package:state_notifier/state_notifier.dart';
@@ -671,7 +672,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unread: 0,
       readActivities: readActivities,
-      lastReadAt: DateTime.timestamp(),
+      lastReadAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
@@ -691,7 +692,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unseen: 0,
       seenActivities: seenActivities,
-      lastSeenAt: DateTime.timestamp(),
+      lastSeenAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
@@ -715,7 +716,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unread: updatedUnreadCount,
       readActivities: updatedReadActivities,
-      lastReadAt: DateTime.timestamp(),
+      lastReadAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
@@ -739,7 +740,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unseen: updatedUnseenCount,
       seenActivities: updatedSeenActivities,
-      lastSeenAt: DateTime.timestamp(),
+      lastSeenAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);

packages/stream_feeds/lib/src/utils/batcher.dart

@@ -1,5 +1,7 @@
 import 'dart:async';
 
+import 'package:clock/clock.dart';
+
 class Batcher<T, R> {
   Batcher({
     required this.action,
@@ -23,20 +25,22 @@ class Batcher<T, R> {
     }
 
     _itemsToProcess.add(item);
-    _nextActionCompleter ??= _planBatchFetch();
-    return _nextActionCompleter!.future;
+
+    // `_planBatchFetch` owns this field, and clears it when it runs on the spot.
+    final completer = _nextActionCompleter ?? _planBatchFetch();
+    return completer.future;
   }
 
   void dispose() {
     _timer?.cancel();
   }
 
   Completer<R> _planBatchFetch() {
-    final timeSinceLastRun = DateTime.now().difference(_lastRun);
+    final timeSinceLastRun = clock.now().difference(_lastRun);
 
     final newActionCompleter = Completer<R>();
     _nextActionCompleter = newActionCompleter;
-    _lastRun = DateTime.now();
+    _lastRun = clock.now();
 
     if (timeSinceLastRun >= interval) {
       _runBatch();