fix(mobile): caching thumbnails to disk (#21275)

mobile/lib/infrastructure/loaders/remote_image_request.dart

@@ -65,40 +65,53 @@ class RemoteImageRequest extends ImageRequest {
       return null;
     }
 
-    // Handle unknown content length from reverse proxy
-    final contentLength = response.contentLength;
-    final Uint8List bytes;
-    int offset = 0;
-
-    if (contentLength >= 0) {
-      // Known content length - use pre-allocated buffer
-      bytes = Uint8List(contentLength);
-      final subscription = response.listen((List<int> chunk) {
-        // this is important to break the response stream if the request is cancelled
+    final cacheManager = this.cacheManager;
+    final streamController = StreamController<List<int>>(sync: true);
+    final Stream<List<int>> stream;
+    cacheManager?.putStreamedFile(url, streamController.stream);
+    stream = response.map((chunk) {
       if (_isCancelled) {
         throw StateError('Cancelled request');
       }
+      if (cacheManager != null) {
+        streamController.add(chunk);
+      }
+      return chunk;
+    });
+
+    try {
+      final Uint8List bytes = await _downloadBytes(stream, response.contentLength);
+      streamController.close();
+      return await ImmutableBuffer.fromUint8List(bytes);
+    } catch (e) {
+      streamController.addError(e);
+      streamController.close();
+      if (_isCancelled) {
+        return null;
+      }
+      rethrow;
+    }
+  }
+
+  Future<Uint8List> _downloadBytes(Stream<List<int>> stream, int length) async {
+    final Uint8List bytes;
+    int offset = 0;
+    if (length > 0) {
+      // Known content length - use pre-allocated buffer
+      bytes = Uint8List(length);
+      await stream.listen((chunk) {
         bytes.setAll(offset, chunk);
         offset += chunk.length;
-      }, cancelOnError: true);
-      cacheManager?.putStreamedFile(url, response);
-      await subscription.asFuture();
+      }, cancelOnError: true).asFuture();
     } else {
       // Unknown content length - collect chunks dynamically
       final chunks = <List<int>>[];
       int totalLength = 0;
-      final subscription = response.listen((List<int> chunk) {
-        // this is important to break the response stream if the request is cancelled
-        if (_isCancelled) {
-          throw StateError('Cancelled request');
-        }
+      await stream.listen((chunk) {
         chunks.add(chunk);
         totalLength += chunk.length;
-      }, cancelOnError: true);
-      cacheManager?.putStreamedFile(url, response);
-      await subscription.asFuture();
+      }, cancelOnError: true).asFuture();
 
-      // Combine all chunks into a single buffer
       bytes = Uint8List(totalLength);
       for (final chunk in chunks) {
         bytes.setAll(offset, chunk);
@@ -106,7 +119,7 @@ class RemoteImageRequest extends ImageRequest {
       }
     }
 
-    return await ImmutableBuffer.fromUint8List(bytes);
+    return bytes;
   }
 
   Future<ImageInfo?> _loadCachedFile(

mobile/lib/presentation/widgets/images/thumbnail.widget.dart

@@ -94,7 +94,7 @@ class _ThumbnailState extends State<Thumbnail> with SingleTickerProviderStateMix
           imageInfo.dispose();
           return;
         }
-
+        _fadeController.value = 1.0;
         setState(() {
           _providerImage = imageInfo.image;
         });
@@ -115,7 +115,7 @@ class _ThumbnailState extends State<Thumbnail> with SingleTickerProviderStateMix
     final imageStream = _imageStream = imageProvider.resolve(ImageConfiguration.empty);
     final imageStreamListener = _imageStreamListener = ImageStreamListener(
       (ImageInfo imageInfo, bool synchronousCall) {
-        _stopListeningToStream();
+        _stopListeningToThumbhashStream();
         if (!mounted) {
           imageInfo.dispose();
           return;

mobile/lib/providers/image/cache/remote_image_cache_manager.dart

@@ -38,9 +38,21 @@ abstract class RemoteCacheManager extends CacheManager {
     final file = await store.fileSystem.createFile(path);
     final sink = file.openWrite();
     try {
-      await source.pipe(sink);
+      await source.listen(sink.add, cancelOnError: true).asFuture();
     } catch (e) {
+      try {
         await sink.close();
+        await file.delete();
+      } catch (e) {
+        _log.severe('Failed to delete incomplete cache file: $e');
+      }
+      return;
+    }
+
+    try {
+      await sink.flush();
+      await sink.close();
+    } catch (e) {
       try {
         await file.delete();
       } catch (e) {