From e6ac4e4415cac75fe6cd2effbeeb590024235765 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 17:06:43 +0200 Subject: [PATCH 01/25] feat(llc)!: adopt the stream_core error layer and upload task API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two core changes land together, because the upload rework is stacked on the error layer and needs its exception types. **Errors.** Every failure now arrives as a `StreamException` subclass, so `ClientException`, `HttpClientException` and `StreamApiError` are gone. The API seams adopt `runApiSafely`, which classifies a transport failure at the point it happens rather than leaving a raw `StreamDioException` for callers to unwrap — `capabilities_repository` reads a status code off `StreamApiException` directly instead of digging through two layers. `connect` now throws a `StateError` when it is asked for a connection it already has, keeping exceptions for failures. **Uploads.** `processRequestsBatch` uploads every request's attachments as one batch instead of one batch per request, so `maxConcurrent` bounds the uploads across the whole call. Previously N requests meant N concurrency limits and a failure in one left the others uploading work about to be discarded. `processRequest` is now the single-request case of the same path. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/CHANGELOG.md | 4 +- .../stream_feeds/lib/src/cdn/cdn_api.dart | 2 +- .../lib/src/client/feeds_client_impl.dart | 33 +- .../stream_feeds/lib/src/feeds_client.dart | 11 +- .../src/generated/api/api/default_api.dart | 2 +- .../repository/capabilities_repository.dart | 11 +- .../stream_feeds/lib/src/state/activity.dart | 11 + packages/stream_feeds/lib/src/state/feed.dart | 6 + .../stream_feeds/lib/src/utils/uploader.dart | 182 +++++----- packages/stream_feeds/pubspec.yaml | 2 +- .../test/client/feeds_client_test.dart | 30 +- .../stream_feeds/test/state/feed_test.dart | 269 ++++++++++++++ .../test/utils/uploader_test.dart | 342 ++++++++++++++++++ 14 files changed, 772 insertions(+), 135 deletions(-) create mode 100644 packages/stream_feeds/test/utils/uploader_test.dart diff --git a/melos.yaml b/melos.yaml index abe240de..0bd65213 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: f83b5d4d706a79fc429de2d27aead4394b83c1fb + ref: 6c52495a7a670ee6746d1ca09df72b4f10f8a57f path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/CHANGELOG.md b/packages/stream_feeds/CHANGELOG.md index e8273f1a..d47b8f3d 100644 --- a/packages/stream_feeds/CHANGELOG.md +++ b/packages/stream_feeds/CHANGELOG.md @@ -7,6 +7,7 @@ - `PollResponseData.votingVisibility` is now required, so anything constructing one directly must supply it - `ActivityCommentList.state` returns `ActivityCommentListState` rather than `StateNotifier`, matching the other state classes - Removed the call, recording, streaming and chat types that were never part of the Feeds API +- Every failure now arrives as a `StreamException` subclass — `StreamApiException`, `StreamNetworkException`, `StreamAuthenticationException` or `StreamClientException` — in place of `ClientException`, `HttpClientException` and `StreamApiError`, which are gone. `StreamFeedsException` aliases the base type, so catching it catches all four, and each carries the `cause` it was built from ### ✨ Features @@ -35,9 +36,10 @@ ### 🔄 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 +- `connect` throws a `StateError` when a connection is already established or in progress, and a `StreamFeedsException` carrying the cause when it fails - Renamed the types below. The old names still compile, with a deprecation warning, and `dart fix --apply` migrates them: | Old name | New name | diff --git a/packages/stream_feeds/lib/src/cdn/cdn_api.dart b/packages/stream_feeds/lib/src/cdn/cdn_api.dart index 3953b005..3eabf014 100644 --- a/packages/stream_feeds/lib/src/cdn/cdn_api.dart +++ b/packages/stream_feeds/lib/src/cdn/cdn_api.dart @@ -45,5 +45,5 @@ abstract interface class CdnApi { class _ResultCallAdapter extends CallAdapter, Future>> { @override - Future> adapt(Future Function() call) => runSafely(call); + Future> adapt(Future Function() call) => runApiSafely(call); } diff --git a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart index cb01daf8..3fec5998 100644 --- a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart +++ b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart @@ -255,15 +255,17 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { Future _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 const StreamAuthenticationException( + message: 'The token was refused and the provider has no other to give', + ); } } @@ -297,11 +299,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 +334,17 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { Future _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, + cause: error, stackTrace: stackTrace, - ), - ); + ); + + throw exception; + }); final tokenProvider = TokenProvider.static(response.token); _logger.d(() => 'guest created, server assigned ${response.user.id}'); @@ -370,7 +375,11 @@ 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); + + throw exception; } _logger.d(() => 'connected ${user.id}'); diff --git a/packages/stream_feeds/lib/src/feeds_client.dart b/packages/stream_feeds/lib/src/feeds_client.dart index 947570de..061c1fa8 100644 --- a/packages/stream_feeds/lib/src/feeds_client.dart +++ b/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 every failure 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: /// diff --git a/packages/stream_feeds/lib/src/generated/api/api/default_api.dart b/packages/stream_feeds/lib/src/generated/api/api/default_api.dart index eb735350..f4345545 100644 --- a/packages/stream_feeds/lib/src/generated/api/api/default_api.dart +++ b/packages/stream_feeds/lib/src/generated/api/api/default_api.dart @@ -801,5 +801,5 @@ abstract interface class DefaultApi { class _ResultCallAdapter extends CallAdapter, Future>> { @override - Future> adapt(Future Function() call) => runSafely(call); + Future> adapt(Future Function() call) => runApiSafely(call); } diff --git a/packages/stream_feeds/lib/src/repository/capabilities_repository.dart b/packages/stream_feeds/lib/src/repository/capabilities_repository.dart index d60c3919..974d4c13 100644 --- a/packages/stream_feeds/lib/src/repository/capabilities_repository.dart +++ b/packages/stream_feeds/lib/src/repository/capabilities_repository.dart @@ -83,18 +83,11 @@ extension on Result>> { 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) { + final exception = failure.error; + if (exception is! StreamApiException) { return false; } final statusCode = exception.statusCode; - if (statusCode == null) { - return false; - } return statusCode < 100 || statusCode >= 500; } } diff --git a/packages/stream_feeds/lib/src/state/activity.dart b/packages/stream_feeds/lib/src/state/activity.dart index 295f1e2b..b48b6c12 100644 --- a/packages/stream_feeds/lib/src/state/activity.dart +++ b/packages/stream_feeds/lib/src/state/activity.dart @@ -186,6 +186,12 @@ 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]. + /// /// Returns a [Result] containing the created [CommentData] or an error. Future> addComment({ required ActivityAddCommentRequest request, @@ -203,6 +209,11 @@ 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]. + /// /// Returns a [Result] containing a list of created [CommentData] or an error. Future>> addCommentsBatch( List requests, diff --git a/packages/stream_feeds/lib/src/state/feed.dart b/packages/stream_feeds/lib/src/state/feed.dart index 1b5ac20a..e65833b3 100644 --- a/packages/stream_feeds/lib/src/state/feed.dart +++ b/packages/stream_feeds/lib/src/state/feed.dart @@ -186,6 +186,12 @@ 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]. + /// /// Returns a [Result] containing the added [ActivityData] if successful, or an error if the /// operation fails. Future> addActivity({ diff --git a/packages/stream_feeds/lib/src/utils/uploader.dart b/packages/stream_feeds/lib/src/utils/uploader.dart index 93408cc1..831264a4 100644 --- a/packages/stream_feeds/lib/src/utils/uploader.dart +++ b/packages/stream_feeds/lib/src/utils/uploader.dart @@ -26,118 +26,108 @@ extension HasAttachmentsExtension on StreamAttachmentUploader { /// Returns a [Result] containing the updated request or an error. Future> processRequest>( T request, { - OnBatchUploadProgress? onProgress, int maxConcurrent = 5, bool eagerError = true, }) async { - final attachmentsToUpload = request.attachmentUploads; - // If there are no attachments to upload, return the original request. - if (attachmentsToUpload == null || attachmentsToUpload.isEmpty) { - return Result.success(request); - } - - final uploadResult = await _uploadAll( - attachmentsToUpload, - onProgress: onProgress, + final processed = await processRequestsBatch( + [request], maxConcurrent: maxConcurrent, eagerError: eagerError, ); - return uploadResult.map((attachments) { - final uploadedAttachments = { - for (final uploaded in attachments) - uploaded.id: api.Attachment( - type: uploaded.type, - custom: {...?uploaded.custom}, - assetUrl: uploaded.remoteUrl, - imageUrl: uploaded.remoteUrl, - thumbUrl: uploaded.thumbnailUrl, - ), - }; - - // Merge uploaded attachments with existing ones, avoiding duplicates - final current = request.attachments ?? []; - final updatedAttachments = current.merge( - uploadedAttachments.values, - key: (it) => (it.type, it.assetUrl, it.imageUrl), - ); - - // Remove processed uploads from the upload queue using ID-based filtering - final uploadedIds = uploadedAttachments.keys.toSet(); - final updatedAttachmentUploads = attachmentsToUpload.where( - (upload) => !uploadedIds.contains(upload.id), - ); - - return request.withAttachments( - attachments: updatedAttachments.takeIf((it) => it.isNotEmpty), - attachmentUploads: updatedAttachmentUploads.toList(), - ); - }); + return processed.map((requests) => requests.single); } /// Processes multiple requests with attachment uploads in parallel. /// - /// Processes each request individually using [processRequest] and returns - /// a list of updated requests with all attachments ready for API submission. + /// Uploads every request's attachments as one batch, so [maxConcurrent] + /// bounds the uploads across all of them rather than within each one, and + /// merges each request's own attachments back into it. + /// + /// When [eagerError] is true the first upload to fail becomes the result's + /// and the rest are called off; when false every attachment is attempted and + /// the ones that did not make it stay queued on their own request. /// - /// Returns a [Result] containing the list of updated requests or an error. + /// Throws an [ArgumentError] if two requests share an attachment id, which + /// one batch could not tell apart. Future>> processRequestsBatch>( List requests, { - OnBatchUploadProgress? onProgress, int maxConcurrent = 5, bool eagerError = true, - }) { - return runSafely(() async { - final batch = requests.map( - (request) => processRequest( - request, - onProgress: onProgress, - maxConcurrent: maxConcurrent, - eagerError: eagerError, - ), - ); - - final processed = await batch.wait; - - final successfulRequests = []; - for (final result in processed) { - // If eagerError is enabled, throw on first failure - if (result.exceptionOrNull() case final error? when eagerError) { - final stackTrace = result.stackTraceOrNull(); - Error.throwWithStackTrace(error, stackTrace ?? StackTrace.current); - } - - successfulRequests.add(result.getOrNull()); - } - - return successfulRequests.nonNulls.toList(); - }); - } + }) async { + final attachmentsToUpload = [ + for (final request in requests) ...?request.attachmentUploads, + ]; - // Uploads multiple attachments in parallel with progress tracking. - // - // Processes [attachments] in batches with configurable concurrency and progress - // reporting. Returns a [Result] containing the list of uploaded attachments. - // - // Returns a [Result] containing a list of [UploadedAttachment] or an error. - Future>> _uploadAll( - Iterable attachments, { - OnBatchUploadProgress? onProgress, - int maxConcurrent = 5, - bool eagerError = true, - }) { - return runSafely(() async { - final batch = uploadBatch( - attachments, - onProgress: onProgress, - maxConcurrent: maxConcurrent, - eagerError: eagerError, - ); - - final batchResult = await batch.toList(); - final uploadedAttachments = batchResult.map((it) => it.getOrNull()); - - return uploadedAttachments.nonNulls.toList(); - }); + if (attachmentsToUpload.isEmpty) return Result.success(requests); + + final batch = uploadBatch( + attachmentsToUpload, + maxConcurrent: maxConcurrent, + eagerError: eagerError, + ); + + return switch (await batch.result) { + BatchUploadCompleted(:final items) => Result.success(_distribute(requests, items)), + BatchUploadStoppedOnError(:final error) => Result.failure(error, error.stackTrace), + BatchUploadCancelled() => const Result.failure( + StreamNetworkException(message: 'The attachment uploads were cancelled', isCancelled: true), + ), + }; } } + +// Each request with its share of the batch folded in. +// +// The map is built once and looked up per request, so this stays linear in the +// number of attachments however many requests share the batch. +List _distribute>( + List requests, + List items, +) { + final succeeded = items.map((it) => it.result.getOrNull()).nonNulls; + final uploaded = {for (final it in succeeded) it.id: _toApiAttachment(it)}; + + return [for (final request in requests) _withUploaded(request, uploaded)]; +} + +// An uploaded attachment as the api model a request carries. +// +// Both urls get the same one whatever the attachment is, and `type` is what +// tells them apart. The Swift SDK maps it the same way; Android instead fills +// `imageUrl` for images only. +api.Attachment _toApiAttachment( + UploadedAttachment attachment, +) => api.Attachment( + type: attachment.type, + custom: {...?attachment.custom}, + assetUrl: attachment.remoteUrl, + imageUrl: attachment.remoteUrl, + thumbUrl: attachment.thumbnailUrl, +); + +// One request's share of [uploaded] merged in beside the attachments it already +// had and taken off its upload queue. Whatever did not make it stays queued for +// a later attempt. +T _withUploaded>( + T request, + Map uploaded, +) { + final queued = request.attachmentUploads; + if (queued == null || queued.isEmpty) return request; + + final merged = queued.map((it) => uploaded[it.id]).nonNulls; + final stillQueued = queued.where((it) => !uploaded.containsKey(it.id)); + + // Merge uploaded attachments with existing ones, avoiding duplicates + final current = request.attachments ?? []; + final updated = current.merge( + merged, + key: (it) => (it.type, it.assetUrl, it.imageUrl), + ); + + return request.withAttachments( + attachments: updated.takeIf((it) => it.isNotEmpty), + attachmentUploads: stillQueued.toList(), + ); +} diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 8f7723bb..60d75b98 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: f83b5d4d706a79fc429de2d27aead4394b83c1fb + ref: 6c52495a7a670ee6746d1ca09df72b4f10f8a57f path: packages/stream_core uuid: ^4.5.1 diff --git a/packages/stream_feeds/test/client/feeds_client_test.dart b/packages/stream_feeds/test/client/feeds_client_test.dart index 94432036..a22466fd 100644 --- a/packages/stream_feeds/test/client/feeds_client_test.dart +++ b/packages/stream_feeds/test/client/feeds_client_test.dart @@ -48,7 +48,7 @@ void main() { ); // Attempt connection - should fail - await expectLater(tester.client.connect(), throwsA(isA())); + await expectLater(tester.client.connect(), throwsA(isA())); // Verify state transitions expectation await connectionStateExpectation; @@ -121,7 +121,7 @@ void main() { body: (tester) async { await expectLater( tester.client.connect(), - throwsA(isA().having((it) => it.apiError?.code, 'apiError.code', 40)), + throwsA(isA().having((it) => it.code, 'code', 40)), ); }, ); @@ -229,9 +229,17 @@ void main() { }, body: (tester) async { // Ignored, this would sit in `Authenticating` until the connect timeout swept it up. + // Nothing authenticated the credentials and nothing refused them, so the failure + // reported is the send's own rather than a verdict on the token. await expectLater( tester.client.connect(), - throwsA(isA()), + throwsA( + isA().having( + (it) => it.cause, + 'cause', + isA().having((it) => '$it', 'message', contains('mid-handshake')), + ), + ), ); expect( @@ -256,8 +264,8 @@ void main() { await expectLater( tester.client.connect(), throwsA( - isA().having( - (it) => it.underlyingError, + isA().having( + (it) => it.cause, 'cause', isA().having((it) => '$it', 'message', contains('token endpoint is down')), ), @@ -330,7 +338,7 @@ void main() { // Told they asked for something they already have, rather than silently doing nothing. expect( () => tester.client.connect(), - throwsA(isA().having((it) => it.message, 'message', contains('already available'))), + throwsA(isA().having((it) => it.message, 'message', contains('already available'))), ); // The connection it already had is left alone. @@ -350,7 +358,7 @@ void main() { expect( () => tester.client.connect(), - throwsA(isA().having((it) => it.message, 'message', contains('already in progress'))), + throwsA(isA().having((it) => it.message, 'message', contains('already in progress'))), ); // The attempt already under way is the one that completes. @@ -451,7 +459,7 @@ void main() { await tester.client.disconnect(); // Reported, rather than left waiting on a connection no longer coming. - await expectLater(connecting, throwsA(isA())); + await expectLater(connecting, throwsA(isA())); expect(tester.client.connectionState.value, isA()); }, ); @@ -1398,9 +1406,9 @@ void main() { await expectLater( tester.client.connect(), throwsA( - isA() + isA() .having((it) => it.message, 'message', 'Failed to create a guest user') - .having((it) => it.underlyingError, 'cause', isException), + .having((it) => it.cause, 'cause', isException), ), ); @@ -1498,7 +1506,7 @@ void main() { error: Exception('Failed to create guest'), ); - await expectLater(tester.client.connect(), throwsA(isA())); + await expectLater(tester.client.connect(), throwsA(isA())); addTearDown(tester.client.dispose); }, diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart index af3ebeae..f73a7c98 100644 --- a/packages/stream_feeds/test/state/feed_test.dart +++ b/packages/stream_feeds/test/state/feed_test.dart @@ -1,5 +1,7 @@ // ignore_for_file: avoid_redundant_argument_values +import 'dart:io'; + import 'package:stream_feeds/stream_feeds.dart'; import 'package:stream_feeds_test/stream_feeds_test.dart'; @@ -4918,4 +4920,271 @@ void main() { ), ); }); + + group('Attachment uploads', () { + const feedId = FeedId(group: 'user', id: 'john'); + + const uploadResponse = FileUploadResponse( + duration: '0ms', + file: 'https://cdn/a', + thumbUrl: 'https://cdn/a-thumb', + ); + + const refused = StreamApiException( + message: 'too large', + statusCode: 413, + code: StreamErrorCode.payloadTooBig, + ); + + // What the file upload above merges into, once posted. + const uploadedAttachment = Attachment( + custom: {}, + type: 'file', + assetUrl: 'https://cdn/a', + imageUrl: 'https://cdn/a', + thumbUrl: 'https://cdn/a-thumb', + ); + + const mergedRequest = AddActivityRequest( + type: 'post', + feeds: [], + attachments: [uploadedAttachment], + ); + + const mergedMixedRequest = AddActivityRequest( + type: 'post', + feeds: [], + attachments: [ + uploadedAttachment, + Attachment( + custom: {}, + type: 'image', + assetUrl: 'https://cdn/img', + imageUrl: 'https://cdn/img', + ), + ], + ); + + const mergedCommentRequest = AddCommentRequest( + comment: 'Look at this', + objectId: 'activity-1', + objectType: 'activity', + attachments: [uploadedAttachment], + ); + + const mergedCommentsBatchRequest = AddCommentsBatchRequest( + comments: [ + AddCommentRequest( + comment: 'with attachment', + objectId: 'activity-1', + objectType: 'activity', + attachments: [uploadedAttachment], + ), + AddCommentRequest( + comment: 'plain', + objectId: 'activity-1', + objectType: 'activity', + ), + ], + ); + + StreamAttachment uploadAttachment(String id, {AttachmentType type = AttachmentType.file}) { + // A real file: the CDN client reads the upload's bytes from its path. + final file = File('${Directory.systemTemp.createTempSync('feeds_upload_test').path}/$id.bin') + ..writeAsBytesSync(const [1, 2, 3]); + + return StreamAttachment(id: id, type: type, file: AttachmentFile(file.path)); + } + + FeedAddActivityRequest requestWithUpload() { + return FeedAddActivityRequest( + type: 'post', + attachmentUploads: [uploadAttachment('a')], + ); + } + + setUpAll(() { + registerFallbackValue([]); + registerFallbackValue(const AddActivityRequest(type: 'post', feeds: [])); + registerFallbackValue(const AddCommentRequest(comment: 'fallback', objectId: 'x', objectType: 'activity')); + }); + + Future> uploadCall(CdnApi cdn) => cdn.uploadFile( + file: any(named: 'file'), + onUploadProgress: any(named: 'onUploadProgress'), + cancelToken: any(named: 'cancelToken'), + ); + + Future> imageUploadCall(CdnApi cdn) => cdn.uploadImage( + file: any(named: 'file'), + onUploadProgress: any(named: 'onUploadProgress'), + cancelToken: any(named: 'cancelToken'), + ); + + feedTest( + 'addActivity() - uploads the attachments and posts the merged request', + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + tester.mockCdn(uploadCall, result: uploadResponse); + tester.mockApi( + (api) => api.addActivity(addActivityRequest: mergedRequest), + result: AddActivityResponse( + duration: '0ms', + activity: createDefaultActivityResponse( + id: 'activity-1', + feeds: [feedId.rawValue], + ), + ), + ); + + final result = await tester.feed.addActivity(request: requestWithUpload()); + + expect(result.isSuccess, isTrue); + }, + // The uploaded file arrived merged into the posted request, its upload + // queue spent. + verify: (tester) => tester.verifyApi( + (api) => api.addActivity(addActivityRequest: mergedRequest), + ), + ); + + feedTest( + "addActivity() - fails with the upload's own error, posting nothing", + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + tester.mockCdnFailure(uploadCall, error: refused); + + final result = await tester.feed.addActivity(request: requestWithUpload()); + + // The upload's failure surfaces classified, and nothing was posted + // for an activity whose attachments never made it. + expect(result.exceptionOrNull(), same(refused)); + tester.verifyNeverCalled( + (api) => api.addActivity(addActivityRequest: any(named: 'addActivityRequest')), + ); + }, + ); + + feedTest( + 'addActivity() - routes each attachment through its own endpoint and merges them all', + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + tester.mockCdn(uploadCall, result: uploadResponse); + tester.mockCdn( + imageUploadCall, + result: const ImageUploadResponse(duration: '0ms', file: 'https://cdn/img'), + ); + tester.mockApi( + (api) => api.addActivity(addActivityRequest: mergedMixedRequest), + result: AddActivityResponse( + duration: '0ms', + activity: createDefaultActivityResponse(id: 'activity-1', feeds: [feedId.rawValue]), + ), + ); + + final result = await tester.feed.addActivity( + request: FeedAddActivityRequest( + type: 'post', + attachmentUploads: [ + uploadAttachment('a'), + uploadAttachment('b', type: AttachmentType.image), + ], + ), + ); + + expect(result.isSuccess, isTrue); + }, + // The file went through the file endpoint, the image through the image + // one, and both arrived merged into the posted request. + verify: (tester) => tester.verifyApi( + (api) => api.addActivity(addActivityRequest: mergedMixedRequest), + ), + ); + + feedTest( + 'addComment() - uploads the attachments and posts the merged comment', + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + tester.mockCdn(uploadCall, result: uploadResponse); + tester.mockApi( + (api) => api.addComment(addCommentRequest: mergedCommentRequest), + result: createDefaultAddCommentResponse(objectId: 'activity-1', text: 'Look at this'), + ); + + final result = await tester.feed.addComment( + request: ActivityAddCommentRequest( + activityId: 'activity-1', + comment: 'Look at this', + attachmentUploads: [uploadAttachment('a')], + ), + ); + + expect(result.isSuccess, isTrue); + }, + // The comment went out with the uploaded attachment merged in. + verify: (tester) => tester.verifyApi( + (api) => api.addComment(addCommentRequest: mergedCommentRequest), + ), + ); + + feedTest( + "addComment() - fails with the upload's own error, posting nothing", + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + tester.mockCdnFailure(uploadCall, error: refused); + + final result = await tester.feed.addComment( + request: ActivityAddCommentRequest( + activityId: 'activity-1', + comment: 'Look at this', + attachmentUploads: [uploadAttachment('a')], + ), + ); + + expect(result.exceptionOrNull(), same(refused)); + tester.verifyNeverCalled( + (api) => api.addComment(addCommentRequest: any(named: 'addCommentRequest')), + ); + }, + ); + + activityTest( + "addCommentsBatch() - uploads each request's attachments before posting the batch", + build: (client) => client.activity(activityId: 'activity-1', fid: feedId), + body: (tester) async { + tester.mockCdn(uploadCall, result: uploadResponse); + tester.mockApi( + (api) => api.addCommentsBatch(addCommentsBatchRequest: mergedCommentsBatchRequest), + result: AddCommentsBatchResponse( + duration: '0ms', + comments: [ + createDefaultCommentResponse(id: 'comment-1', objectId: 'activity-1', text: 'with attachment'), + createDefaultCommentResponse(id: 'comment-2', objectId: 'activity-1', text: 'plain'), + ], + ), + ); + + final result = await tester.activity.addCommentsBatch([ + ActivityAddCommentRequest( + activityId: 'activity-1', + comment: 'with attachment', + attachmentUploads: [uploadAttachment('a')], + ), + const ActivityAddCommentRequest(activityId: 'activity-1', comment: 'plain'), + ]); + + expect(result.isSuccess, isTrue); + }, + // Each comment keeps its own attachments: the upload landed on the + // request that carried it, and only on that one. + verify: (tester) => tester.verifyApi( + (api) => api.addCommentsBatch(addCommentsBatchRequest: mergedCommentsBatchRequest), + ), + ); + }); } diff --git a/packages/stream_feeds/test/utils/uploader_test.dart b/packages/stream_feeds/test/utils/uploader_test.dart new file mode 100644 index 00000000..55a4924e --- /dev/null +++ b/packages/stream_feeds/test/utils/uploader_test.dart @@ -0,0 +1,342 @@ +import 'dart:async'; + +import 'package:stream_core/stream_core.dart'; +import 'package:stream_feeds/src/generated/api/models.dart' as api; +import 'package:stream_feeds/src/utils/uploader.dart'; +import 'package:stream_feeds_test/stream_feeds_test.dart'; + +class _TestRequest implements HasAttachments<_TestRequest> { + const _TestRequest({this.attachments, this.attachmentUploads}); + + @override + final List? attachments; + + @override + final List? attachmentUploads; + + @override + _TestRequest withAttachments({ + List? attachments, + List? attachmentUploads, + }) => _TestRequest(attachments: attachments, attachmentUploads: attachmentUploads); +} + +/// A CDN whose outcome per file is scripted by [outcomes]. +class _FakeCdn implements CdnClient { + _FakeCdn(this.outcomes); + + final Map> outcomes; + + Future> _upload(AttachmentFile file) async => outcomes[file]!; + + @override + Future> uploadFile( + AttachmentFile file, { + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) => _upload(file); + + @override + Future> uploadImage( + AttachmentFile image, { + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) => _upload(image); + + @override + Future> deleteFile(String url, {CancelToken? cancelToken}) async => const Result.success(null); + + @override + Future> deleteImage(String url, {CancelToken? cancelToken}) async => const Result.success(null); +} + +/// A CDN that parks every upload until the test answers it, so how many are on +/// the wire at once is observable. +class _GatedCdn implements CdnClient { + final _pending = >>{}; + + /// How many uploads are on the wire right now. + int get inFlight => _pending.values.where((it) => !it.isCompleted).length; + + /// Answers the upload of [file] with [outcome]. + void answer(AttachmentFile file, Result outcome) => _pending[file]!.complete(outcome); + + Future> _upload(AttachmentFile file) => + (_pending[file] ??= Completer>()).future; + + @override + Future> uploadFile( + AttachmentFile file, { + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) => _upload(file); + + @override + Future> uploadImage( + AttachmentFile image, { + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) => _upload(image); + + @override + Future> deleteFile(String url, {CancelToken? cancelToken}) async => const Result.success(null); + + @override + Future> deleteImage(String url, {CancelToken? cancelToken}) async => const Result.success(null); +} + +StreamAttachment _attachment(String id, AttachmentFile file) => StreamAttachment( + id: id, + type: AttachmentType.file, + file: file, +); + +const _refused = StreamApiException(message: 'too large', statusCode: 413, code: StreamErrorCode.payloadTooBig); + +void main() { + final fileA = AttachmentFile.fromData(Uint8List(0)); + final fileB = AttachmentFile.fromData(Uint8List(0)); + + const uploadedA = UploadedFile(fileUrl: 'https://cdn/a', thumbUrl: 'https://cdn/a-thumb'); + const uploadedB = UploadedFile(fileUrl: 'https://cdn/b'); + + group('processRequest', () { + test('returns the request untouched when there is nothing to upload', () async { + final uploader = StreamAttachmentUploader(cdn: _FakeCdn({})); + const request = _TestRequest(attachmentUploads: []); + + final result = await uploader.processRequest(request); + + expect(result.getOrNull(), same(request)); + }); + + test('merges every uploaded attachment and empties the upload queue', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}), + ); + final request = _TestRequest( + attachmentUploads: [_attachment('a', fileA), _attachment('b', fileB)], + ); + + final processed = await uploader.processRequest(request); + + final updated = processed.getOrNull()!; + expect( + updated.attachments?.map((it) => it.assetUrl), + unorderedEquals(['https://cdn/a', 'https://cdn/b']), + ); + expect( + updated.attachments?.map((it) => it.thumbUrl), + unorderedEquals(['https://cdn/a-thumb', null]), + ); + expect(updated.attachmentUploads, isEmpty); + }); + + test('keeps existing attachments and does not duplicate an already merged upload', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA)}), + ); + const existing = api.Attachment(custom: {}, assetUrl: 'https://cdn/existing'); + const alreadyMerged = api.Attachment( + custom: {}, + type: 'file', + assetUrl: 'https://cdn/a', + imageUrl: 'https://cdn/a', + ); + final request = _TestRequest( + attachments: const [existing, alreadyMerged], + attachmentUploads: [_attachment('a', fileA)], + ); + + final processed = await uploader.processRequest(request); + + expect( + processed.getOrNull()!.attachments?.map((it) => it.assetUrl), + unorderedEquals(['https://cdn/existing', 'https://cdn/a']), + ); + }); + + test("fails as one when an upload fails, carrying the upload's own error", () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}), + ); + final request = _TestRequest( + attachmentUploads: [_attachment('a', fileA), _attachment('b', fileB)], + ); + + final processed = await uploader.processRequest(request); + + expect(processed.exceptionOrNull(), same(_refused)); + }); + + test('without eagerError, keeps the failed upload queued for a later attempt', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}), + ); + final request = _TestRequest( + attachmentUploads: [_attachment('a', fileA), _attachment('b', fileB)], + ); + + final processed = await uploader.processRequest(request, eagerError: false); + + final updated = processed.getOrNull()!; + expect(updated.attachments?.map((it) => it.assetUrl), ['https://cdn/a']); + expect(updated.attachmentUploads?.map((it) => it.id), ['b']); + }); + }); + + group('processRequestsBatch', () { + test('gives each request only its own uploaded attachments', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}), + ); + final first = _TestRequest(attachmentUploads: [_attachment('a', fileA)]); + final second = _TestRequest(attachmentUploads: [_attachment('b', fileB)]); + + final processed = await uploader.processRequestsBatch([first, second]); + final requests = processed.getOrNull()!; + + expect(requests.first.attachments?.map((it) => it.assetUrl), ['https://cdn/a']); + expect(requests.last.attachments?.map((it) => it.assetUrl), ['https://cdn/b']); + expect(requests.every((it) => it.attachmentUploads!.isEmpty), isTrue); + }); + + test('leaves a request whose upload failed queued without touching the others', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}), + ); + final ok = _TestRequest(attachmentUploads: [_attachment('a', fileA)]); + final failed = _TestRequest(attachmentUploads: [_attachment('b', fileB)]); + + final processed = await uploader.processRequestsBatch([ok, failed], eagerError: false); + final requests = processed.getOrNull()!; + + expect(requests.first.attachments?.map((it) => it.assetUrl), ['https://cdn/a']); + expect(requests.first.attachmentUploads, isEmpty); + expect(requests.last.attachments, isNull); + expect(requests.last.attachmentUploads?.map((it) => it.id), ['b']); + }); + + test('returns the requests in the order they were given', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}), + ); + final withB = _TestRequest(attachmentUploads: [_attachment('b', fileB)]); + final withA = _TestRequest(attachmentUploads: [_attachment('a', fileA)]); + + final processed = await uploader.processRequestsBatch([withB, withA]); + final requests = processed.getOrNull()!; + + expect(requests.first.attachments?.map((it) => it.assetUrl), ['https://cdn/b']); + expect(requests.last.attachments?.map((it) => it.assetUrl), ['https://cdn/a']); + }); + + test('leaves a request with nothing to upload exactly as it was', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA)}), + ); + const nothingToUpload = _TestRequest(); + final withUpload = _TestRequest(attachmentUploads: [_attachment('a', fileA)]); + + final processed = await uploader.processRequestsBatch([nothingToUpload, withUpload]); + final requests = processed.getOrNull()!; + + expect(requests.first, same(nothingToUpload), reason: 'nothing about it changed'); + expect(requests.last.attachments?.map((it) => it.assetUrl), ['https://cdn/a']); + }); + + test('bounds the uploads across every request, not within each one', () async { + final cdn = _GatedCdn(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final files = [for (var i = 0; i < 4; i++) AttachmentFile.fromData(Uint8List(0))]; + final requests = <_TestRequest>[]; + for (var i = 0; i < files.length; i++) { + requests.add(_TestRequest(attachmentUploads: [_attachment('a$i', files[i])])); + } + + final pending = uploader.processRequestsBatch(requests, maxConcurrent: 2); + await pumpEventQueue(); + + expect(cdn.inFlight, 2, reason: 'four requests of one attachment each, two uploads at a time'); + + cdn.answer(files[0], const Result.success(uploadedA)); + await pumpEventQueue(); + + expect(cdn.inFlight, 2, reason: 'a freed slot takes exactly one more'); + + for (final file in files.skip(1)) { + cdn.answer(file, const Result.success(uploadedA)); + await pumpEventQueue(); + } + + expect((await pending).getOrNull(), hasLength(4)); + }); + + test('refuses two requests that share an attachment id', () async { + final uploader = StreamAttachmentUploader(cdn: _FakeCdn({})); + final shared = _attachment('same', fileA); + + await expectLater( + uploader.processRequestsBatch([ + _TestRequest(attachmentUploads: [shared]), + _TestRequest(attachmentUploads: [shared]), + ]), + throwsArgumentError, + reason: 'one batch could not tell them apart', + ); + }); + + test('succeeds with nothing when there are no requests', () async { + final uploader = StreamAttachmentUploader(cdn: _FakeCdn({})); + + final processed = await uploader.processRequestsBatch<_TestRequest>([]); + + expect(processed.getOrNull(), isEmpty); + }); + + test('processes every request when all uploads succeed', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}), + ); + final requests = [ + _TestRequest(attachmentUploads: [_attachment('a', fileA)]), + _TestRequest(attachmentUploads: [_attachment('b', fileB)]), + ]; + + final processed = await uploader.processRequestsBatch(requests); + + expect(processed.getOrNull(), hasLength(2)); + }); + + test('fails as one when any request fails', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}), + ); + final requests = [ + _TestRequest(attachmentUploads: [_attachment('a', fileA)]), + _TestRequest(attachmentUploads: [_attachment('b', fileB)]), + ]; + + final processed = await uploader.processRequestsBatch(requests); + + expect(processed.exceptionOrNull(), same(_refused)); + }); + + test('without eagerError, keeps every request, the failed upload staying queued in its own', () async { + final uploader = StreamAttachmentUploader( + cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}), + ); + final requests = [ + _TestRequest(attachmentUploads: [_attachment('a', fileA)]), + _TestRequest(attachmentUploads: [_attachment('b', fileB)]), + ]; + + final processed = await uploader.processRequestsBatch(requests, eagerError: false); + + final kept = processed.getOrNull()!; + expect(kept, hasLength(2)); + expect(kept.first.attachmentUploads, isEmpty); + expect(kept.last.attachmentUploads?.map((it) => it.id), ['b']); + }); + }); +} From 745f2486f4a59253d8d454455e15a87038ce688a Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 17:21:29 +0200 Subject: [PATCH 02/25] chore(deps): re-pin stream_core to the reviewed upload API Picks up the review fixes on GetStream/stream-core-flutter#170, including `UploadProgress.totalBytes` becoming nullable. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index 0bd65213..8e5a5f08 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 6c52495a7a670ee6746d1ca09df72b4f10f8a57f + ref: 09345039ad65c0bb014909aafe9cda3ec7b65c61 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 60d75b98..903dec2e 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 6c52495a7a670ee6746d1ca09df72b4f10f8a57f + ref: 09345039ad65c0bb014909aafe9cda3ec7b65c61 path: packages/stream_core uuid: ^4.5.1 From d4c870e40b9e4bcd73069950594e89e3c7a3dc79 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 17:26:03 +0200 Subject: [PATCH 03/25] chore(deps): re-pin stream_core to the batch constructor cleanup Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index 8e5a5f08..703991f6 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 09345039ad65c0bb014909aafe9cda3ec7b65c61 + ref: 16f77227333fd334d7f37c78c9a80caa5195e1b6 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 903dec2e..6c75ec81 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 09345039ad65c0bb014909aafe9cda3ec7b65c61 + ref: 16f77227333fd334d7f37c78c9a80caa5195e1b6 path: packages/stream_core uuid: ^4.5.1 From a02c45ebc13796004dc2a10b3cb807368b8f25b9 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 17:55:37 +0200 Subject: [PATCH 04/25] chore(deps): re-pin stream_core, and import dart:typed_data directly The pinned commit stops re-exporting `dart:typed_data` from the core barrel, which is where `uploader_test` was reaching `Uint8List`. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- packages/stream_feeds/test/utils/uploader_test.dart | 1 + 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index 703991f6..c0a744a0 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 16f77227333fd334d7f37c78c9a80caa5195e1b6 + ref: 02e86e7ecca5ca14bb913f12168f9eb392e30dfe path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 6c75ec81..ef469a36 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 16f77227333fd334d7f37c78c9a80caa5195e1b6 + ref: 02e86e7ecca5ca14bb913f12168f9eb392e30dfe path: packages/stream_core uuid: ^4.5.1 diff --git a/packages/stream_feeds/test/utils/uploader_test.dart b/packages/stream_feeds/test/utils/uploader_test.dart index 55a4924e..fc55f71f 100644 --- a/packages/stream_feeds/test/utils/uploader_test.dart +++ b/packages/stream_feeds/test/utils/uploader_test.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:typed_data'; import 'package:stream_core/stream_core.dart'; import 'package:stream_feeds/src/generated/api/models.dart' as api; From 39c109bb570bec221e58f7c57da1494b3cd943da Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 18:10:14 +0200 Subject: [PATCH 05/25] docs(llc): update the upload snippet to the task API `upload` hands back a running `AttachmentUploadTask` rather than a future, so the snippet no longer awaits it, drops the `onProgress` callback that went with the old signature, and reads progress off the task's state instead. Progress is shown the way the API asks for: the fraction is absent until the file's length is known. Co-Authored-By: Claude Opus 5 (1M context) --- docs/code_snippets/03_03_file_uploads.dart | 23 ++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/docs/code_snippets/03_03_file_uploads.dart b/docs/code_snippets/03_03_file_uploads.dart index 9e443045..bc37245b 100644 --- a/docs/code_snippets/03_03_file_uploads.dart +++ b/docs/code_snippets/03_03_file_uploads.dart @@ -20,14 +20,21 @@ Future 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 - }, - ); + // Start the upload. The task comes back straight away, already running. + 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'); + } + }); + + // Await the outcome. It never throws — a failed upload carries its error. + // Call `task.cancel()` to call the upload off. + final result = await task.result; // Map the result to an Attachment model to send with an activity final uploadedAttachment = result.getOrThrow(); From e5f858e87b6f941f7b3db6d49d0b53717529142f Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 18:16:25 +0200 Subject: [PATCH 06/25] fix(llc): act on review of the error layer migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The exception raised when a static provider's token is refused dropped the refusal itself — the `StreamApiException` that ended the session was right there and went nowhere. It is the `cause` now, with its stack. `StreamFeedsException` was documented as the root of *every* failure the SDK reports, two lines above a `connect` that throws `StateError` for being called twice. Both it and the changelog entry now say what they mean: failures for work the SDK attempted. Asking for something it cannot be asked stays a programming error. The upload test helper left a directory in the system temp folder per attachment, on every run. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/CHANGELOG.md | 2 +- .../stream_feeds/lib/src/client/feeds_client_impl.dart | 4 +++- packages/stream_feeds/lib/src/feeds_client.dart | 7 +++++-- packages/stream_feeds/test/state/feed_test.dart | 8 +++++--- 4 files changed, 14 insertions(+), 7 deletions(-) diff --git a/packages/stream_feeds/CHANGELOG.md b/packages/stream_feeds/CHANGELOG.md index d47b8f3d..3e360e4b 100644 --- a/packages/stream_feeds/CHANGELOG.md +++ b/packages/stream_feeds/CHANGELOG.md @@ -7,7 +7,7 @@ - `PollResponseData.votingVisibility` is now required, so anything constructing one directly must supply it - `ActivityCommentList.state` returns `ActivityCommentListState` rather than `StateNotifier`, matching the other state classes - Removed the call, recording, streaming and chat types that were never part of the Feeds API -- Every failure now arrives as a `StreamException` subclass — `StreamApiException`, `StreamNetworkException`, `StreamAuthenticationException` or `StreamClientException` — in place of `ClientException`, `HttpClientException` and `StreamApiError`, which are gone. `StreamFeedsException` aliases the base type, so catching it catches all four, and each carries the `cause` it was built from +- Every failure the SDK reports for work it attempted now arrives as a `StreamException` subclass — `StreamApiException`, `StreamNetworkException`, `StreamAuthenticationException` or `StreamClientException` — in place of `ClientException`, `HttpClientException` and `StreamApiError`, which are gone. `StreamFeedsException` aliases the base type, so catching it catches all four, and each carries the `cause` it was built from ### ✨ Features diff --git a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart index 3fec5998..4f948dea 100644 --- a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart +++ b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart @@ -263,8 +263,10 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { // 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 const StreamAuthenticationException( + throw StreamAuthenticationException( message: 'The token was refused and the provider has no other to give', + cause: previousError, + stackTrace: previousError?.stackTrace, ); } } diff --git a/packages/stream_feeds/lib/src/feeds_client.dart b/packages/stream_feeds/lib/src/feeds_client.dart index 061c1fa8..4870b45b 100644 --- a/packages/stream_feeds/lib/src/feeds_client.dart +++ b/packages/stream_feeds/lib/src/feeds_client.dart @@ -49,10 +49,13 @@ import 'state/user_list.dart'; export 'client/moderation_client.dart'; -/// The root of every failure this SDK reports. +/// The root of the failures this SDK reports for work it attempted. /// /// An alias of [StreamException], so code written against either name catches -/// the same failures. +/// the same failures. Asking the client for something it cannot be asked — +/// connecting twice, or using it after [StreamFeedsClient.dispose] — is a +/// programming error and throws a [StateError] instead, which is not meant to +/// be caught. typedef StreamFeedsException = StreamException; /// {@template stream_feeds_client} diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart index f73a7c98..05a14473 100644 --- a/packages/stream_feeds/test/state/feed_test.dart +++ b/packages/stream_feeds/test/state/feed_test.dart @@ -4990,10 +4990,12 @@ void main() { StreamAttachment uploadAttachment(String id, {AttachmentType type = AttachmentType.file}) { // A real file: the CDN client reads the upload's bytes from its path. - final file = File('${Directory.systemTemp.createTempSync('feeds_upload_test').path}/$id.bin') - ..writeAsBytesSync(const [1, 2, 3]); + final directory = Directory.systemTemp.createTempSync('feeds_upload_test'); + addTearDown(() => directory.deleteSync(recursive: true)); - return StreamAttachment(id: id, type: type, file: AttachmentFile(file.path)); + File('${directory.path}/$id.bin').writeAsBytesSync(const [1, 2, 3]); + + return StreamAttachment(id: id, type: type, file: AttachmentFile('${directory.path}/$id.bin')); } FeedAddActivityRequest requestWithUpload() { From 06196811d5a03eb1583081b7d418706475271e47 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 18:27:46 +0200 Subject: [PATCH 07/25] docs(llc): say that colliding attachment ids throw MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `addActivity`, `addComment` and `addCommentsBatch` return a `Result`, which reads as a promise not to throw, and then throw an `ArgumentError` when two attachments in one upload share an id. That is the right behaviour — misuse is not a condition to handle, so it stays out of the `Result` — but nothing said it happens. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/lib/src/state/activity.dart | 9 +++++++++ packages/stream_feeds/lib/src/state/feed.dart | 4 ++++ 2 files changed, 13 insertions(+) diff --git a/packages/stream_feeds/lib/src/state/activity.dart b/packages/stream_feeds/lib/src/state/activity.dart index b48b6c12..784fbf6a 100644 --- a/packages/stream_feeds/lib/src/state/activity.dart +++ b/packages/stream_feeds/lib/src/state/activity.dart @@ -192,6 +192,10 @@ class Activity with Disposable { /// `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> addComment({ required ActivityAddCommentRequest request, @@ -214,6 +218,11 @@ class Activity with Disposable { /// 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>> addCommentsBatch( List requests, diff --git a/packages/stream_feeds/lib/src/state/feed.dart b/packages/stream_feeds/lib/src/state/feed.dart index e65833b3..7ba8f109 100644 --- a/packages/stream_feeds/lib/src/state/feed.dart +++ b/packages/stream_feeds/lib/src/state/feed.dart @@ -192,6 +192,10 @@ class Feed with Disposable { /// `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> addActivity({ From 13a4e250ee6276799c6cf3a3a6a50bc7986bb357 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 18:44:46 +0200 Subject: [PATCH 08/25] docs(llc): shorten the StreamFeedsException doc back down The `StateError` paragraph restated what `connect` already documents at the point a caller meets it, so it earned nothing here. What the review was right about was one word: "every failure" claimed more than the alias covers. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/lib/src/feeds_client.dart | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/packages/stream_feeds/lib/src/feeds_client.dart b/packages/stream_feeds/lib/src/feeds_client.dart index 4870b45b..df06417c 100644 --- a/packages/stream_feeds/lib/src/feeds_client.dart +++ b/packages/stream_feeds/lib/src/feeds_client.dart @@ -49,13 +49,10 @@ import 'state/user_list.dart'; export 'client/moderation_client.dart'; -/// The root of the failures this SDK reports for work it attempted. +/// The root of the failures this SDK reports. /// /// An alias of [StreamException], so code written against either name catches -/// the same failures. Asking the client for something it cannot be asked — -/// connecting twice, or using it after [StreamFeedsClient.dispose] — is a -/// programming error and throws a [StateError] instead, which is not meant to -/// be caught. +/// the same failures. typedef StreamFeedsException = StreamException; /// {@template stream_feeds_client} From 3182f0897f9509494e28dfbe6e97f3252811b29e Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 18:49:15 +0200 Subject: [PATCH 09/25] docs(llc): stop the upload snippet contradicting itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit It said the outcome never throws, one line above a `getOrThrow()` that rethrows the carried error. The comment now says what `getOrThrow` is for — opting into a throw — and points at `fold` for handling it instead. Co-Authored-By: Claude Opus 5 (1M context) --- docs/code_snippets/03_03_file_uploads.dart | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/docs/code_snippets/03_03_file_uploads.dart b/docs/code_snippets/03_03_file_uploads.dart index bc37245b..977a2d11 100644 --- a/docs/code_snippets/03_03_file_uploads.dart +++ b/docs/code_snippets/03_03_file_uploads.dart @@ -20,7 +20,8 @@ Future howToUploadAFileOrImageStep1() async { custom: {'width': 600, 'height': 400}, ); - // Start the upload. The task comes back straight away, already running. + // 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 @@ -32,12 +33,10 @@ Future howToUploadAFileOrImageStep1() async { } }); - // Await the outcome. It never throws — a failed upload carries its error. - // Call `task.cancel()` to call the upload off. - final result = await task.result; - - // Map the result to an Attachment model to send with an activity - final uploadedAttachment = result.getOrThrow(); + // 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, From c4414b9405077d57bfbdcabac4f9fe57bc3987cc Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 18:57:58 +0200 Subject: [PATCH 10/25] chore(deps): re-pin stream_core to the doc cleanup Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index c0a744a0..da0f63a9 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 02e86e7ecca5ca14bb913f12168f9eb392e30dfe + ref: c7366d2a82157b577080d8a81d887a85a5b60211 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index ef469a36..2d7a12fb 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 02e86e7ecca5ca14bb913f12168f9eb392e30dfe + ref: c7366d2a82157b577080d8a81d887a85a5b60211 path: packages/stream_core uuid: ^4.5.1 From 20d6b6fc76c382e656ec586d02dec7167a36dce1 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 19:05:24 +0200 Subject: [PATCH 11/25] docs(changelog): file the connect throw under breaking, not changed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 0.5.1 had no already-established guard on `connect`, so code that called it twice went on to try again; it now throws a `StateError`. That breaks callers, and the policy reserves `🔄 Changed` for what does not. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/stream_feeds/CHANGELOG.md b/packages/stream_feeds/CHANGELOG.md index 3e360e4b..3b4f158a 100644 --- a/packages/stream_feeds/CHANGELOG.md +++ b/packages/stream_feeds/CHANGELOG.md @@ -7,7 +7,8 @@ - `PollResponseData.votingVisibility` is now required, so anything constructing one directly must supply it - `ActivityCommentList.state` returns `ActivityCommentListState` rather than `StateNotifier`, 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` — in place of `ClientException`, `HttpClientException` and `StreamApiError`, which are gone. `StreamFeedsException` aliases the base type, so catching it catches all four, and each carries the `cause` it was built from +- Every failure the SDK reports for work it attempted now arrives as a `StreamException` subclass — `StreamApiException`, `StreamNetworkException`, `StreamAuthenticationException` or `StreamClientException` — replacing `ClientException`, `HttpClientException` and `StreamApiError`. `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 ### ✨ Features @@ -39,7 +40,6 @@ - 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 `StateError` when a connection is already established or in progress, and a `StreamFeedsException` carrying the cause when it fails - Renamed the types below. The old names still compile, with a deprecation warning, and `dart fix --apply` migrates them: | Old name | New name | From e382757db8d0fb11eda27fc33859d7d0c2e9b5ab Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 19:14:14 +0200 Subject: [PATCH 12/25] chore(deps): re-pin stream_core, dropping the typed_data import again `stream_core` re-exports `Uint8List` once more, so the explicit import here is redundant and the analyzer says so. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- packages/stream_feeds/test/utils/uploader_test.dart | 1 - 3 files changed, 2 insertions(+), 3 deletions(-) diff --git a/melos.yaml b/melos.yaml index da0f63a9..58b02541 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: c7366d2a82157b577080d8a81d887a85a5b60211 + ref: f78d72f95b147fefdc06e6f6a5d269183e98aa55 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 2d7a12fb..f421d479 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: c7366d2a82157b577080d8a81d887a85a5b60211 + ref: f78d72f95b147fefdc06e6f6a5d269183e98aa55 path: packages/stream_core uuid: ^4.5.1 diff --git a/packages/stream_feeds/test/utils/uploader_test.dart b/packages/stream_feeds/test/utils/uploader_test.dart index fc55f71f..55a4924e 100644 --- a/packages/stream_feeds/test/utils/uploader_test.dart +++ b/packages/stream_feeds/test/utils/uploader_test.dart @@ -1,5 +1,4 @@ import 'dart:async'; -import 'dart:typed_data'; import 'package:stream_core/stream_core.dart'; import 'package:stream_feeds/src/generated/api/models.dart' as api; From 309fe20a78cb10e61d5070f0fc71c04d297091ad Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 19:18:25 +0200 Subject: [PATCH 13/25] chore(deps): re-pin stream_core to the restored typed_data export Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index 58b02541..f9f52390 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: f78d72f95b147fefdc06e6f6a5d269183e98aa55 + ref: fd4611e8140608ff23fc5a36b9d426fe672ea303 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index f421d479..441b09dd 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: f78d72f95b147fefdc06e6f6a5d269183e98aa55 + ref: fd4611e8140608ff23fc5a36b9d426fe672ea303 path: packages/stream_core uuid: ^4.5.1 From b79c29a5fba9bbc02c0e4b006d88518b1783a649 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 19:42:33 +0200 Subject: [PATCH 14/25] fix(llc): retry a capabilities fetch that failed on the network MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The predicate only retried a `StreamApiException`, so a request that never reached the server — a timeout or an offline blip, which arrives as `StreamNetworkException` — was dropped on the first attempt. That is the likeliest failure for a mobile client and the one `ERROR_LAYER.md`'s retry table names for a read. Rate limits stay out, and the backend is why: `/feeds/own/batch` sits on the tighter limit tier and its 429 carries no `Retry-After`, so a fixed 500 ms retry would spend the budget rather than wait out the limit. Nothing covered this path before — no test in the package touched `ownBatch`. The new one drives it the way a consumer does, through an activity event naming an uncached feed, and fails on the old behaviour with 1 call instead of 2. `createDefaultActivityResponse` gained a `currentFeed` parameter to make that reachable. Co-Authored-By: Claude Opus 5 (1M context) --- .../repository/capabilities_repository.dart | 25 ++++++------ .../stream_feeds/test/state/feed_test.dart | 38 +++++++++++++++++++ .../lib/src/helpers/test_data.dart | 2 + 3 files changed, 51 insertions(+), 14 deletions(-) diff --git a/packages/stream_feeds/lib/src/repository/capabilities_repository.dart b/packages/stream_feeds/lib/src/repository/capabilities_repository.dart index 974d4c13..29177697 100644 --- a/packages/stream_feeds/lib/src/repository/capabilities_repository.dart +++ b/packages/stream_feeds/lib/src/repository/capabilities_repository.dart @@ -77,18 +77,15 @@ class CapabilitiesRepository { } extension on Result>> { - bool shouldRetry() { - switch (this) { - case api.Success(): - return false; - - case final api.Failure failure: - final exception = failure.error; - if (exception is! StreamApiException) { - return false; - } - final statusCode = exception.statusCode; - 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, + }, + }; } diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart index 05a14473..333ffa02 100644 --- a/packages/stream_feeds/test/state/feed_test.dart +++ b/packages/stream_feeds/test/state/feed_test.dart @@ -666,6 +666,44 @@ void main() { }, ); + feedTest( + 'ActivityAddedEvent - retries a capabilities fetch that failed on the network', + user: const User(id: 'user-1'), + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + registerFallbackValue(const OwnBatchRequest(feeds: [])); + + // The activity names a feed this client has not cached, which is what + // sends the handler to fetch its capabilities. + tester.mockApiFailure( + (api) => api.ownBatch(ownBatchRequest: any(named: 'ownBatchRequest')), + error: const StreamNetworkException(message: 'Connection failed'), + ); + + await tester.emitEvent( + ActivityAddedEvent( + type: EventTypes.activityAdded, + createdAt: DateTime.timestamp(), + custom: const {}, + fid: feedId.rawValue, + activity: createDefaultActivityResponse( + id: 'new-activity', + userId: 'user-1', + currentFeed: createDefaultFeedResponse(id: 'other', groupId: 'user'), + ), + ), + ); + + // A blip is worth asking again for; the loop allows exactly one retry. + await Future.delayed(const Duration(milliseconds: 700)); + tester.verifyApiCalled( + (api) => api.ownBatch(ownBatchRequest: any(named: 'ownBatchRequest')), + times: 2, + ); + }, + ); + feedTest( 'ActivityAddedEvent - should add activity to feed', user: const User(id: 'user-1'), diff --git a/packages/stream_feeds_test/lib/src/helpers/test_data.dart b/packages/stream_feeds_test/lib/src/helpers/test_data.dart index 583b5569..cecddd0a 100644 --- a/packages/stream_feeds_test/lib/src/helpers/test_data.dart +++ b/packages/stream_feeds_test/lib/src/helpers/test_data.dart @@ -103,6 +103,7 @@ ActivityResponse createDefaultActivityResponse({ int? friendReactionCount, Map? metrics, ActivityResponseRestrictReplies? restrictReplies, + FeedResponse? currentFeed, }) { latestReactions = latestReactions.isEmpty ? ownReactions : latestReactions; reactionGroups = switch (reactionGroups.isNotEmpty) { @@ -157,6 +158,7 @@ ActivityResponse createDefaultActivityResponse({ reactionCount: reactionGroups.values.sumOf((group) => group.count), reactionGroups: reactionGroups, restrictReplies: restrictReplies ?? ActivityResponseRestrictReplies.everyone, + currentFeed: currentFeed, score: 0, searchData: const {}, shareCount: 0, From 240d4259571901a5feb790b1d61e5a64fdce7dd9 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Mon, 31 Aug 2026 21:08:45 +0200 Subject: [PATCH 15/25] chore(deps): re-pin stream_core to the reconnection doc fixes Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index f9f52390..0cefc979 100644 --- a/melos.yaml +++ b/melos.yaml @@ -53,7 +53,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: fd4611e8140608ff23fc5a36b9d426fe672ea303 + ref: ca4f472846592f7221e9cede6ec0adf89846e39e path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 441b09dd..b8d43d73 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -41,7 +41,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: fd4611e8140608ff23fc5a36b9d426fe672ea303 + ref: ca4f472846592f7221e9cede6ec0adf89846e39e path: packages/stream_core uuid: ^4.5.1 From 3f787f9069de4b64aa279f663b6b4a1fb738d761 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 14:18:45 +0200 Subject: [PATCH 16/25] docs(llc): say what the error layer removed, and what else it broke MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `StreamApiError` is not gone: it is still the server's error payload and still the type of `ConnectionErrorEvent.error`. What changed is that it is no longer what the SDK throws or returns, so someone matching on `connection.error` was being sent looking for a replacement that does not exist. Two breaking changes were missing entirely. The attachment uploader's `upload` and `uploadBatch` changed shape, and they are public here through `StreamFeedsClient.attachmentUploader`. And `addActivity`, `addComment` and `addCommentsBatch` now throw on a duplicate attachment id where they used to report through the `Result` — a change no compiler will point out. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/stream_feeds/CHANGELOG.md b/packages/stream_feeds/CHANGELOG.md index 3b4f158a..39a67f28 100644 --- a/packages/stream_feeds/CHANGELOG.md +++ b/packages/stream_feeds/CHANGELOG.md @@ -7,8 +7,10 @@ - `PollResponseData.votingVisibility` is now required, so anything constructing one directly must supply it - `ActivityCommentList.state` returns `ActivityCommentListState` rather than `StateNotifier`, 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`, `HttpClientException` and `StreamApiError`. `StreamFeedsException` aliases the base type, so one `on` clause catches all four +- 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>`, and takes no `onProgress`: progress arrives on the task's `state`. `uploadBatch` returns an `AttachmentUploadBatch` rather than a `Stream>` +- `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 From 60f0caf44c14444ec4cea84a46d48cfc29024cf4 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 14:18:45 +0200 Subject: [PATCH 17/25] fix(llc): keep the stack trace that points at the failure Rethrowing a classified exception plainly restarts its trace at the rethrow, so what reached the caller pointed at `feeds_client_impl` rather than at the request or the socket that actually failed. `Result.getOrThrow` in core already uses `Error.throwWithStackTrace` for this reason. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/lib/src/client/feeds_client_impl.dart | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart index 4f948dea..e9b90c8e 100644 --- a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart +++ b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart @@ -345,7 +345,7 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { stackTrace: stackTrace, ); - throw exception; + Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current); }); final tokenProvider = TokenProvider.static(response.token); @@ -381,7 +381,7 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { var exception = StreamException.tryFrom(source.cause); exception ??= StreamNetworkException(message: source.closeReason, cause: source.cause); - throw exception; + Error.throwWithStackTrace(exception, exception.stackTrace ?? StackTrace.current); } _logger.d(() => 'connected ${user.id}'); From 0e4477343f8ec19ee48ec7f8f2f698df469eae4d Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 14:18:46 +0200 Subject: [PATCH 18/25] docs(test): name the exceptions connect actually throws Four examples still told readers to expect a `ClientException`, which this branch removes. The types are the ones the client tests assert: a refused token is answered by the server, so `StreamApiException`; credentials that could not be sent never reach it, so `StreamClientException`. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds_test/lib/src/testers/base_tester.dart | 4 ++-- .../stream_feeds_test/lib/src/testers/websocket_tester.dart | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/stream_feeds_test/lib/src/testers/base_tester.dart b/packages/stream_feeds_test/lib/src/testers/base_tester.dart index 04860786..34ce2db5 100644 --- a/packages/stream_feeds_test/lib/src/testers/base_tester.dart +++ b/packages/stream_feeds_test/lib/src/testers/base_tester.dart @@ -107,7 +107,7 @@ abstract base class BaseTester with ApiMockerMixin, CdnMockerMixin { /// // Custom error code (triggers reconnection) /// tester.mockFailedAuth(errorCode: 5); /// - /// await expectLater(client.connect(), throwsA(isA())); + /// await expectLater(client.connect(), throwsA(isA())); /// ``` void mockFailedAuth({int errorCode = 40}) { return _wsTester.mockFailedAuth(errorCode: errorCode); @@ -122,7 +122,7 @@ abstract base class BaseTester with ApiMockerMixin, CdnMockerMixin { /// ```dart /// tester.mockFailedSend(); /// - /// await expectLater(client.connect(), throwsA(isA())); + /// await expectLater(client.connect(), throwsA(isA())); /// ``` void mockFailedSend({Object? error}) { return _wsTester.mockFailedSend(error: error); diff --git a/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart b/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart index 8b637921..ccbe0281 100644 --- a/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart +++ b/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart @@ -90,7 +90,7 @@ final class WebSocketTester { /// Example: /// ```dart /// wsTester.mockFailedSend(); - /// await expectLater(client.connect(), throwsA(isA())); + /// await expectLater(client.connect(), throwsA(isA())); /// ``` void mockFailedSend({Object? error}) { _resetFunction?.call(); // Reset previous mocks if any @@ -137,7 +137,7 @@ final class WebSocketTester { /// /// await expectLater( /// client.connect(), - /// throwsA(isA()), + /// throwsA(isA()), /// ); /// ``` void mockFailedAuth({int errorCode = 40}) { From d77277c5ae60182651916d08c8ece4dc139b3b9d Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 14:19:09 +0200 Subject: [PATCH 19/25] test(llc): ask for the capabilities request the handler actually sends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `any(named: 'ownBatchRequest')` matched anything, so the test passed whatever feed the handler asked about. Naming the request pins it — and shows the call goes out as `OwnBatchRequest(feeds: ['other'])`, the bare feed id rather than the fid. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_feeds/test/state/feed_test.dart | 41 +++++++++++-------- 1 file changed, 23 insertions(+), 18 deletions(-) diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart index 333ffa02..4ffa4a1b 100644 --- a/packages/stream_feeds/test/state/feed_test.dart +++ b/packages/stream_feeds/test/state/feed_test.dart @@ -666,6 +666,25 @@ void main() { }, ); + // The activity below names a feed this client has not cached, which is what + // sends the handler to fetch its capabilities. + OwnBatchRequest capabilitiesFor(String feed) => OwnBatchRequest(feeds: [feed]); + + // The wait `CapabilitiesRepository` allows itself before its one retry. + const retryBackoff = Duration(milliseconds: 500); + + ActivityAddedEvent activityInUncachedFeed() => ActivityAddedEvent( + type: EventTypes.activityAdded, + createdAt: DateTime.timestamp(), + custom: const {}, + fid: feedId.rawValue, + activity: createDefaultActivityResponse( + id: 'new-activity', + userId: 'user-1', + currentFeed: createDefaultFeedResponse(id: 'other', groupId: 'user'), + ), + ); + feedTest( 'ActivityAddedEvent - retries a capabilities fetch that failed on the network', user: const User(id: 'user-1'), @@ -674,31 +693,17 @@ void main() { body: (tester) async { registerFallbackValue(const OwnBatchRequest(feeds: [])); - // The activity names a feed this client has not cached, which is what - // sends the handler to fetch its capabilities. tester.mockApiFailure( - (api) => api.ownBatch(ownBatchRequest: any(named: 'ownBatchRequest')), + (api) => api.ownBatch(ownBatchRequest: capabilitiesFor('other')), error: const StreamNetworkException(message: 'Connection failed'), ); - await tester.emitEvent( - ActivityAddedEvent( - type: EventTypes.activityAdded, - createdAt: DateTime.timestamp(), - custom: const {}, - fid: feedId.rawValue, - activity: createDefaultActivityResponse( - id: 'new-activity', - userId: 'user-1', - currentFeed: createDefaultFeedResponse(id: 'other', groupId: 'user'), - ), - ), - ); + await tester.emitEvent(activityInUncachedFeed()); // A blip is worth asking again for; the loop allows exactly one retry. - await Future.delayed(const Duration(milliseconds: 700)); + await Future.delayed(retryBackoff + const Duration(milliseconds: 200)); tester.verifyApiCalled( - (api) => api.ownBatch(ownBatchRequest: any(named: 'ownBatchRequest')), + (api) => api.ownBatch(ownBatchRequest: capabilitiesFor('other')), times: 2, ); }, From 89b5f9a4a739da30b748ca401842a0561cfabd52 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 14:19:09 +0200 Subject: [PATCH 20/25] test(llc): pin the duplicate attachment id where callers meet it Two attachments sharing an id throw an `ArgumentError` out of a method that otherwise reports through a `Result`, so a caller that only folds never sees it. That was covered at the `processRequestsBatch` seam; this covers it at `addActivity`, where the contract is actually met, and checks nothing is posted. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_feeds/test/state/feed_test.dart | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart index 4ffa4a1b..15e01f56 100644 --- a/packages/stream_feeds/test/state/feed_test.dart +++ b/packages/stream_feeds/test/state/feed_test.dart @@ -1,5 +1,6 @@ // ignore_for_file: avoid_redundant_argument_values +import 'dart:async'; import 'dart:io'; import 'package:stream_feeds/stream_feeds.dart'; @@ -673,6 +674,7 @@ void main() { // The wait `CapabilitiesRepository` allows itself before its one retry. const retryBackoff = Duration(milliseconds: 500); + ActivityAddedEvent activityInUncachedFeed() => ActivityAddedEvent( type: EventTypes.activityAdded, createdAt: DateTime.timestamp(), @@ -5176,6 +5178,29 @@ void main() { ), ); + feedTest( + 'addActivity() - refuses two attachments that share an id, before uploading anything', + build: (client) => client.feedFromId(feedId), + setUp: (tester) => tester.getOrCreate(), + body: (tester) async { + // A misuse rather than a failure, so it escapes the `Result` the method + // otherwise reports through — a caller that only folds cannot see it. + await expectLater( + tester.feed.addActivity( + request: FeedAddActivityRequest( + type: 'post', + attachmentUploads: [uploadAttachment('same'), uploadAttachment('same')], + ), + ), + throwsArgumentError, + ); + + tester.verifyNeverCalled( + (api) => api.addActivity(addActivityRequest: any(named: 'addActivityRequest')), + ); + }, + ); + feedTest( "addComment() - fails with the upload's own error, posting nothing", build: (client) => client.feedFromId(feedId), From 4f9821f236eaa085495c86fbd102f802188b97cd Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 15:15:39 +0200 Subject: [PATCH 21/25] refactor(llc): read the time through package:clock `DateTime.now()` and `DateTime.timestamp()` read the wall clock, which no test can move. `Batcher` computes its collection window from one, so the window was unobservable; `FeedState`'s read and seen timestamps are stamped from the other. `package:clock` is what `fakeAsync` fakes, so reading through it makes both answerable to a test without changing what they do at runtime. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 ++ packages/stream_feeds/lib/src/state/feed_state.dart | 9 +++++---- packages/stream_feeds/lib/src/utils/batcher.dart | 7 +++++-- packages/stream_feeds/pubspec.yaml | 2 ++ 4 files changed, 14 insertions(+), 6 deletions(-) diff --git a/melos.yaml b/melos.yaml index 0cefc979..bcc86529 100644 --- a/melos.yaml +++ b/melos.yaml @@ -25,6 +25,7 @@ command: auto_route: ^11.0.0 collection: ^1.18.0 chewie: ^1.11.3 + clock: ^1.1.3 dio: ^5.9.0 equatable: ^2.0.5 flutter_state_notifier: ^1.0.0 @@ -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 diff --git a/packages/stream_feeds/lib/src/state/feed_state.dart b/packages/stream_feeds/lib/src/state/feed_state.dart index ed0c1bbd..eecee6ae 100644 --- a/packages/stream_feeds/lib/src/state/feed_state.dart +++ b/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); diff --git a/packages/stream_feeds/lib/src/utils/batcher.dart b/packages/stream_feeds/lib/src/utils/batcher.dart index b8c855aa..f322914c 100644 --- a/packages/stream_feeds/lib/src/utils/batcher.dart +++ b/packages/stream_feeds/lib/src/utils/batcher.dart @@ -1,5 +1,7 @@ import 'dart:async'; +import 'package:clock/clock.dart'; + class Batcher { Batcher({ required this.action, @@ -23,6 +25,7 @@ class Batcher { } _itemsToProcess.add(item); + _nextActionCompleter ??= _planBatchFetch(); return _nextActionCompleter!.future; } @@ -32,11 +35,11 @@ class Batcher { } Completer _planBatchFetch() { - final timeSinceLastRun = DateTime.now().difference(_lastRun); + final timeSinceLastRun = clock.now().difference(_lastRun); final newActionCompleter = Completer(); _nextActionCompleter = newActionCompleter; - _lastRun = DateTime.now(); + _lastRun = clock.now(); if (timeSinceLastRun >= interval) { _runBatch(); diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index b8d43d73..acf8bc33 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -19,6 +19,7 @@ environment: sdk: ^3.12.0 dependencies: + clock: ^1.1.3 collection: ^1.18.0 dio: ^5.9.0 equatable: ^2.0.5 @@ -47,6 +48,7 @@ dependencies: dev_dependencies: build_runner: ^2.4.15 + fake_async: ^1.3.3 freezed: ^3.0.0 json_serializable: ^6.9.5 retrofit_generator: ^10.2.6 From 26e038838b2c9c6a6550c27352be60b9cc56c8f8 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 15:16:02 +0200 Subject: [PATCH 22/25] fix(llc): let a batcher run more than one batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_nextActionCompleter ??= _planBatchFetch()` null-checks the field, then writes the callee's return value back over it — but `_planBatchFetch` owns that field, and clears it when it runs the batch on the spot. So after the first batch the field held a completed completer, every later add joined it instead of planning, and its item sat in `_itemsToProcess` unsent. Feed capabilities go through this, so they were fetched once per client and every feed discovered afterwards was answered with the first batch's result. Six tests cover it, including the two the dartdoc already promised: an add is answered by the batch that carried it, and everyone in a batch is answered with the whole batch. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_feeds/CHANGELOG.md | 1 + .../stream_feeds/lib/src/utils/batcher.dart | 5 +- .../stream_feeds/test/utils/batcher_test.dart | 134 ++++++++++++++++++ 3 files changed, 138 insertions(+), 2 deletions(-) create mode 100644 packages/stream_feeds/test/utils/batcher_test.dart diff --git a/packages/stream_feeds/CHANGELOG.md b/packages/stream_feeds/CHANGELOG.md index 39a67f28..e96ab482 100644 --- a/packages/stream_feeds/CHANGELOG.md +++ b/packages/stream_feeds/CHANGELOG.md @@ -33,6 +33,7 @@ ### 🐛 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 diff --git a/packages/stream_feeds/lib/src/utils/batcher.dart b/packages/stream_feeds/lib/src/utils/batcher.dart index f322914c..9c6360c0 100644 --- a/packages/stream_feeds/lib/src/utils/batcher.dart +++ b/packages/stream_feeds/lib/src/utils/batcher.dart @@ -26,8 +26,9 @@ class Batcher { _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() { diff --git a/packages/stream_feeds/test/utils/batcher_test.dart b/packages/stream_feeds/test/utils/batcher_test.dart new file mode 100644 index 00000000..6c07ad2d --- /dev/null +++ b/packages/stream_feeds/test/utils/batcher_test.dart @@ -0,0 +1,134 @@ +import 'package:fake_async/fake_async.dart'; +import 'package:stream_feeds/src/utils/batcher.dart'; +import 'package:stream_feeds_test/stream_feeds_test.dart'; + +void main() { + group('Batcher', () { + ({Batcher batcher, List> batches}) build({ + Duration interval = const Duration(seconds: 2), + }) { + final batches = >[]; + final batcher = Batcher( + interval: interval, + action: (items) async { + batches.add(items); + return items.length; + }, + ); + + return (batcher: batcher, batches: batches); + } + + test('asks straight away, having nothing to wait behind', () { + fakeAsync((async) { + final (:batcher, :batches) = build(); + + batcher.add('a').ignore(); + async.flushMicrotasks(); + + expect(batches, [ + ['a'], + ]); + }); + }); + + test('starts a new batch once the one before it has run', () { + fakeAsync((async) { + final (:batcher, :batches) = build(); + + batcher.add('a').ignore(); + async.flushMicrotasks(); + + // The second add has to plan a batch of its own: joining the settled + // one would answer it with a result gathered before it asked. + batcher.add('b').ignore(); + async.elapse(const Duration(seconds: 3)); + + expect(batches, [ + ['a'], + ['b'], + ]); + }); + }); + + test('answers each add with the batch that carried it', () { + fakeAsync((async) { + final batcher = Batcher( + action: (items) async => items.join(','), + ); + + String? first; + String? second; + + batcher.add('a').then((it) => first = it).ignore(); + async.flushMicrotasks(); + + batcher.add('b').then((it) => second = it).ignore(); + async.elapse(const Duration(seconds: 3)); + + expect(first, 'a'); + // Joining the settled batch would answer this with 'a' — a result + // gathered before it asked. + expect(second, 'b'); + }); + }); + + test('answers everyone in a batch with the whole batch', () { + fakeAsync((async) { + final batcher = Batcher( + action: (items) async => items.join(','), + ); + final answers = {}; + + batcher.add('a').ignore(); + async.flushMicrotasks(); + + // Both land inside one window, so both are told what the batch as a + // whole came back with, not just their own part of it. + batcher.add('b').then((it) => answers['b'] = it).ignore(); + batcher.add('c').then((it) => answers['c'] = it).ignore(); + async.elapse(const Duration(seconds: 3)); + + expect(answers, {'b': 'b,c', 'c': 'b,c'}); + }); + }); + + test('collects everything that arrives inside one window', () { + fakeAsync((async) { + final (:batcher, :batches) = build(); + + batcher.add('a').ignore(); + async.flushMicrotasks(); + + batcher.add('b').ignore(); + batcher.add('c').ignore(); + async.elapse(const Duration(seconds: 3)); + + expect(batches, [ + ['a'], + ['b', 'c'], + ], reason: 'the two behind the window went out together'); + }); + }); + + test('measures the window against the clock it is given', () { + fakeAsync((async) { + final (:batcher, :batches) = build(); + + batcher.add('a').ignore(); + async.flushMicrotasks(); + + // Reading the wall clock instead would hold this add inside the window + // however long a test waited. + async.elapse(const Duration(seconds: 3)); + batcher.add('b').ignore(); + async.flushMicrotasks(); + + expect(batches, [ + ['a'], + ['b'], + ], reason: 'the window had passed, so it asked on the spot'); + }); + }); + }); +} From c526e98c66a296a8c86f0c77857d23944f0c6bde Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 17:47:43 +0200 Subject: [PATCH 23/25] fix(feeds): follow stream_core dropping stackTrace from its exceptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A `StreamException` no longer carries a trace: the trace describes the raise, so it travels beside the failure. The three sites that read one off an exception now take it from whatever carried it — `BatchUploadStoppedOnError` for a batch that gave up, the `DisconnectionSource` for a connection that closed — or let `throw` capture it where the exception is made here. Also unblocks CI: `clock` is relaxed to `^1.1.2`, matching stream_core, so it resolves against the `clock 1.1.2` the legacy Flutter's `flutter_test` pins. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_feeds/lib/src/client/feeds_client_impl.dart | 9 ++++++--- packages/stream_feeds/lib/src/utils/uploader.dart | 2 +- packages/stream_feeds/pubspec.yaml | 4 ++-- packages/stream_feeds/test/state/feed_test.dart | 1 - 4 files changed, 9 insertions(+), 7 deletions(-) diff --git a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart index e9b90c8e..f40aeeef 100644 --- a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart +++ b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart @@ -266,7 +266,6 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { throw StreamAuthenticationException( message: 'The token was refused and the provider has no other to give', cause: previousError, - stackTrace: previousError?.stackTrace, ); } } @@ -342,7 +341,6 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { exception ??= StreamClientException( message: 'Failed to create a guest user', cause: error, - stackTrace: stackTrace, ); Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current); @@ -378,10 +376,15 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { if (state case Disconnected(:final source)) { _logger.w(() => 'connect ${user.id} failed: ${source.closeReason}', error: source.cause); + final stackTrace = switch (source) { + ServerInitiated(:final stackTrace) || AuthenticationFailed(:final stackTrace) => stackTrace, + UserInitiated() || SystemInitiated() || UnHealthyConnection() || ConnectTimeout() => null, + }; + var exception = StreamException.tryFrom(source.cause); exception ??= StreamNetworkException(message: source.closeReason, cause: source.cause); - Error.throwWithStackTrace(exception, exception.stackTrace ?? StackTrace.current); + Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current); } _logger.d(() => 'connected ${user.id}'); diff --git a/packages/stream_feeds/lib/src/utils/uploader.dart b/packages/stream_feeds/lib/src/utils/uploader.dart index 831264a4..014eea92 100644 --- a/packages/stream_feeds/lib/src/utils/uploader.dart +++ b/packages/stream_feeds/lib/src/utils/uploader.dart @@ -69,7 +69,7 @@ extension HasAttachmentsExtension on StreamAttachmentUploader { return switch (await batch.result) { BatchUploadCompleted(:final items) => Result.success(_distribute(requests, items)), - BatchUploadStoppedOnError(:final error) => Result.failure(error, error.stackTrace), + BatchUploadStoppedOnError(:final error, :final stackTrace) => Result.failure(error, stackTrace), BatchUploadCancelled() => const Result.failure( StreamNetworkException(message: 'The attachment uploads were cancelled', isCancelled: true), ), diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index acf8bc33..665fe863 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -19,7 +19,7 @@ environment: sdk: ^3.12.0 dependencies: - clock: ^1.1.3 + clock: ^1.1.2 collection: ^1.18.0 dio: ^5.9.0 equatable: ^2.0.5 @@ -42,7 +42,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: ca4f472846592f7221e9cede6ec0adf89846e39e + ref: 40b1180f6b5f9ca53a5ecb2c74152fb1c6b4a376 path: packages/stream_core uuid: ^4.5.1 diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart index 15e01f56..d7358b46 100644 --- a/packages/stream_feeds/test/state/feed_test.dart +++ b/packages/stream_feeds/test/state/feed_test.dart @@ -674,7 +674,6 @@ void main() { // The wait `CapabilitiesRepository` allows itself before its one retry. const retryBackoff = Duration(milliseconds: 500); - ActivityAddedEvent activityInUncachedFeed() => ActivityAddedEvent( type: EventTypes.activityAdded, createdAt: DateTime.timestamp(), From 35a902789c24a12a3102c0d56367b6b1da3c9f6d Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Tue, 1 Sep 2026 17:51:21 +0200 Subject: [PATCH 24/25] fix(feeds): point melos at the core ref and clock version the packages use MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `melos.yaml` is where these are declared; bootstrap writes them into every package pubspec, so a stale entry there reverted the pin and stripped the `# ignore: invalid_dependency` comment along with it — which is what failed both analyze jobs. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 4 ++-- packages/stream_feeds/lib/src/client/feeds_client_impl.dart | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/melos.yaml b/melos.yaml index bcc86529..953ed4b2 100644 --- a/melos.yaml +++ b/melos.yaml @@ -25,7 +25,7 @@ command: auto_route: ^11.0.0 collection: ^1.18.0 chewie: ^1.11.3 - clock: ^1.1.3 + clock: ^1.1.2 dio: ^5.9.0 equatable: ^2.0.5 flutter_state_notifier: ^1.0.0 @@ -54,7 +54,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: ca4f472846592f7221e9cede6ec0adf89846e39e + ref: 40b1180f6b5f9ca53a5ecb2c74152fb1c6b4a376 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart index f40aeeef..4d8f213e 100644 --- a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart +++ b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart @@ -376,14 +376,14 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient { if (state case Disconnected(:final source)) { _logger.w(() => 'connect ${user.id} failed: ${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, }; - var exception = StreamException.tryFrom(source.cause); - exception ??= StreamNetworkException(message: source.closeReason, cause: source.cause); - Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current); } From dcee75be9b62663d30e483eb93398414fd2d2c2a Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 11:04:53 +0200 Subject: [PATCH 25/25] chore(feeds): pin stream_core to main now that both core PRs have landed The error layer and the upload task API are on `main` as 505432f; the ref this was pinned to was a branch tip that no longer exists. `packages/stream_core/lib` is byte-identical between the two, so nothing here changes. Co-Authored-By: Claude Opus 5 (1M context) --- melos.yaml | 2 +- packages/stream_feeds/pubspec.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/melos.yaml b/melos.yaml index 953ed4b2..040035c0 100644 --- a/melos.yaml +++ b/melos.yaml @@ -54,7 +54,7 @@ command: stream_core: git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 40b1180f6b5f9ca53a5ecb2c74152fb1c6b4a376 + ref: 505432f213af8911d54ab3c855d6ad6fbe492fc9 path: packages/stream_core video_player: ^2.10.0 uuid: ^4.5.1 diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml index 665fe863..d160bb77 100644 --- a/packages/stream_feeds/pubspec.yaml +++ b/packages/stream_feeds/pubspec.yaml @@ -42,7 +42,7 @@ dependencies: # ignore: invalid_dependency git: url: https://github.com/GetStream/stream-core-flutter.git - ref: 40b1180f6b5f9ca53a5ecb2c74152fb1c6b4a376 + ref: 505432f213af8911d54ab3c855d6ad6fbe492fc9 path: packages/stream_core uuid: ^4.5.1