Prioritize requested attachment downloads

lib/app/layouts/conversation_view/widgets/message/attachment/attachment_holder.dart

@@ -104,15 +104,28 @@ class _AttachmentHolderState extends CustomState<AttachmentHolder, void, Message
         child: InkWell(
           onTap: content is PlatformFile ? null : () async {
             if (content is Attachment && message.error == 0 && !message.guid!.contains("temp")) {
+              final downloader = attachmentDownloader.startDownload(
+                content,
+                onComplete: onComplete,
+                prioritized: true,
+              );
               setState(() {
-                content = attachmentDownloader.startDownload(content, onComplete: onComplete);
+                content = downloader;
               });
             } else if (content is AttachmentDownloadController) {
               final AttachmentDownloadController _content = content;
-              if (!_content.error.value) return;
+              if (!_content.error.value) {
+                attachmentDownloader.prioritize(_content);
+                return;
+              }
               Get.delete<AttachmentDownloadController>(tag: _content.attachment.guid);
+              final downloader = attachmentDownloader.startDownload(
+                _content.attachment,
+                onComplete: onComplete,
+                prioritized: true,
+              );
               setState(() {
-                content = attachmentDownloader.startDownload(_content.attachment, onComplete: onComplete);
+                content = downloader;
               });
             }
           },

lib/services/network/attachment_download_queue.dart

@@ -0,0 +1,49 @@
+class AttachmentDownloadQueue<T> {
+  final Map<String, List<T>> _queues = <String, List<T>>{};
+
+  Iterable<T> get all => _queues.values.expand((items) => items);
+
+  List<T> forChat(String chatGuid) =>
+      List<T>.unmodifiable(_queues[chatGuid] ?? <T>[]);
+
+  void add(String chatGuid, T item, {bool prioritized = false}) {
+    final queue = _queues.putIfAbsent(chatGuid, () => <T>[]);
+    if (queue.contains(item)) {
+      if (prioritized) prioritize(chatGuid, item);
+      return;
+    }
+    if (prioritized) {
+      queue.insert(0, item);
+    } else {
+      queue.add(item);
+    }
+  }
+
+  bool remove(String chatGuid, T item) {
+    final queue = _queues[chatGuid];
+    if (queue == null || !queue.remove(item)) return false;
+    if (queue.isEmpty) _queues.remove(chatGuid);
+    return true;
+  }
+
+  bool prioritize(String chatGuid, T item) {
+    final queue = _queues[chatGuid];
+    if (queue == null || !queue.remove(item)) return false;
+    queue.insert(0, item);
+    return true;
+  }
+
+  T? next({String? activeChatGuid, required bool Function(T) isFetching}) {
+    final activeQueue = activeChatGuid == null ? null : _queues[activeChatGuid];
+    if (activeQueue != null) {
+      for (final item in activeQueue) {
+        if (!isFetching(item)) return item;
+      }
+    }
+
+    for (final item in all) {
+      if (!isFetching(item)) return item;
+    }
+    return null;
+  }
+}

lib/services/network/downloads_service.dart

@@ -1,11 +1,10 @@
 import 'package:bluebubbles/services/network/backend_service.dart';
-import 'package:bluebubbles/utils/file_utils.dart';
+import 'package:bluebubbles/services/network/attachment_download_queue.dart';
 import 'package:bluebubbles/utils/logger/logger.dart';
 import 'package:bluebubbles/helpers/helpers.dart';
 import 'package:bluebubbles/database/models.dart';
 import 'package:bluebubbles/services/services.dart';
 import 'package:collection/collection.dart';
-import 'package:dio/dio.dart';
 import 'package:flutter/foundation.dart';
 import 'package:get/get.dart' hide Response;
 import 'package:path/path.dart';
@@ -18,53 +17,63 @@ AttachmentDownloadService attachmentDownloader = Get.isRegistered<AttachmentDown
 class AttachmentDownloadService extends GetxService {
   int maxDownloads = 2;
   final RxList<String> downloaders = <String>[].obs;
-  final Map<String, List<AttachmentDownloadController>> _downloaders = {};
+  final AttachmentDownloadQueue<AttachmentDownloadController> _downloaders =
+      AttachmentDownloadQueue<AttachmentDownloadController>();
 
   AttachmentDownloadController? getController(String? guid) {
-    return _downloaders.values.flattened.firstWhereOrNull((element) => element.attachment.guid == guid);
+    return _downloaders.all.firstWhereOrNull((element) => element.attachment.guid == guid);
   }
 
-  AttachmentDownloadController startDownload(Attachment a, {Function(PlatformFile)? onComplete, Function? onError}) {
+  AttachmentDownloadController startDownload(Attachment a,
+      {Function(PlatformFile)? onComplete, Function? onError, bool prioritized = false}) {
+    final existing = getController(a.guid);
+    if (existing != null) {
+      if (onComplete != null && !existing.completeFuncs.contains(onComplete)) {
+        existing.completeFuncs.add(onComplete);
+      }
+      if (onError != null && !existing.errorFuncs.contains(onError)) {
+        existing.errorFuncs.add(onError);
+      }
+      if (prioritized && !existing.isFetching) prioritize(existing);
+      return existing;
+    }
     return Get.put(AttachmentDownloadController(
       attachment: a,
       onComplete: onComplete,
       onError: onError,
+      prioritized: prioritized,
     ), tag: a.guid!);
   }
 
   void _addToQueue(AttachmentDownloadController downloader) {
     downloaders.add(downloader.attachment.guid!);
     final chatGuid = downloader.attachment.message.target?.chat.target?.guid ?? "unknown";
-    if (_downloaders.containsKey(chatGuid)) {
-      _downloaders[chatGuid]!.add(downloader);
-    } else {
-      _downloaders[chatGuid] = [downloader];
-    }
+    _downloaders.add(chatGuid, downloader, prioritized: downloader.prioritized);
     _fetchNext();
   }
 
   void _removeFromQueue(AttachmentDownloadController downloader) {
     downloaders.remove(downloader.attachment.guid!);
     final chatGuid = downloader.attachment.message.target?.chat.target?.guid ?? "unknown";
-    _downloaders[chatGuid]!.removeWhere((e) => e.attachment.guid == downloader.attachment.guid);
-    if (_downloaders[chatGuid]!.isEmpty) _downloaders.remove(chatGuid);
+    _downloaders.remove(chatGuid, downloader);
     Get.delete<AttachmentDownloadController>(tag: downloader.attachment.guid!);
     _fetchNext();
   }
 
+  void prioritize(AttachmentDownloadController downloader) {
+    if (downloader.isFetching) return;
+    final chatGuid = downloader.attachment.message.target?.chat.target?.guid ?? "unknown";
+    _downloaders.prioritize(chatGuid, downloader);
+    _fetchNext();
+  }
+
   void _fetchNext() {
-    if (_downloaders.values.flattened.where((e) => e.isFetching).length < maxDownloads) {
-      AttachmentDownloadController? activeChatDownloader;
-      // first check if we have an active chat that needs downloads, if so prioritize that chat
-      if (cm.activeChat != null && _downloaders.containsKey(cm.activeChat!.chat.guid)) {
-        activeChatDownloader = _downloaders[cm.activeChat!.chat.guid]!.firstWhereOrNull((e) => !e.isFetching);
-        activeChatDownloader?.fetchAttachment();
-      }
-      // otherwise just grab a random attachment that needs fetching
-      if (activeChatDownloader == null) {
-        _downloaders.values.flattened.firstWhereOrNull((e) => !e.isFetching)?.fetchAttachment();
-      }
-    }
+    if (_downloaders.all.where((e) => e.isFetching).length >= maxDownloads) return;
+    final next = _downloaders.next(
+      activeChatGuid: cm.activeChat?.chat.guid,
+      isFetching: (downloader) => downloader.isFetching,
+    );
+    next?.fetchAttachment();
   }
 }
 
@@ -75,13 +84,15 @@ class AttachmentDownloadController extends GetxController {
   final RxnNum progress = RxnNum();
   final Rxn<PlatformFile> file = Rxn<PlatformFile>();
   final RxBool error = RxBool(false);
+  final bool prioritized;
   Stopwatch stopwatch = Stopwatch();
   bool isFetching = false;
 
   AttachmentDownloadController({
     required this.attachment,
     Function(PlatformFile)? onComplete,
     Function? onError,
+    this.prioritized = false,
   }) {
     if (onComplete != null) completeFuncs.add(onComplete);
     if (onError != null) errorFuncs.add(onError);

rust/src/api/api.rs

@@ -1895,11 +1895,11 @@ pub async fn download_attachment(sink: StreamSink<TransferProgress>, aps: &APSCo
         let mut file = std::fs::File::create(path)?;
         attachment.get_attachment(aps, &mut file, |prog, total| {
             println!("donwloading file {} of {}", prog, total);
-            sink.add(TransferProgress {
+            let _ = sink.add(TransferProgress {
                 prog,
                 total,
                 attachment: None
-            }).unwrap();
+            });
         }).await?;
         file.flush()?;
         Ok(())
@@ -1914,11 +1914,11 @@ pub async fn download_mmcs(sink: StreamSink<TransferProgress>, aps: &APSConnecti
 
         let mut file = std::fs::File::create(path)?;
         attachment.get_attachment(aps, &mut file, |prog, total| {
-            sink.add(TransferProgress {
+            let _ = sink.add(TransferProgress {
                 prog,
                 total,
                 attachment: None
-            }).unwrap();
+            });
         }).await?;
         file.flush()?;
         Ok(())
@@ -1928,7 +1928,7 @@ pub async fn download_mmcs(sink: StreamSink<TransferProgress>, aps: &APSConnecti
 async fn wrap_sink<Fut, T: SseEncode + Send + Sync>(sink: &StreamSink<T>, f: impl FnOnce() -> Fut)
     where Fut: Future<Output = anyhow::Result<()>> {
     if let Err(err) = f().await {
-        sink.add_error(err).unwrap();
+        let _ = sink.add_error(err);
     }
 }
 
@@ -1945,13 +1945,13 @@ pub async fn upload_mmcs(sink: StreamSink<MMCSTransferProgress>, aps: &APSConnec
         let prepared = MMCSFile::prepare_put(&mut file).await?;
         file.rewind()?;
         let attachment = MMCSFile::new(aps, &prepared, file, |prog, total| {
-            sink.add(MMCSTransferProgress {
+            let _ = sink.add(MMCSTransferProgress {
                 prog,
                 total,
                 file: None
-            }).unwrap();
+            });
         }).await?;
-        sink.add(MMCSTransferProgress { prog: 0, total: 0, file: Some(attachment) }).unwrap();
+        let _ = sink.add(MMCSTransferProgress { prog: 0, total: 0, file: Some(attachment) });
         Ok(())
     }).await
 }
@@ -1963,13 +1963,13 @@ pub async fn upload_attachment(sink: StreamSink<TransferProgress>, aps: &APSConn
         let prepared = MMCSFile::prepare_put(&mut file).await?;
         file.rewind()?;
         let attachment = Attachment::new_mmcs(aps, &prepared, file, &mime, &uti, &name,|prog, total| {
-            sink.add(TransferProgress {
+            let _ = sink.add(TransferProgress {
                 prog,
                 total,
                 attachment: None
-            }).unwrap();
+            });
         }).await?;
-        sink.add(TransferProgress { prog: 0, total: 0, attachment: Some(attachment) }).unwrap();
+        let _ = sink.add(TransferProgress { prog: 0, total: 0, attachment: Some(attachment) });
         Ok(())
     }).await
 }

test/services/network/attachment_download_queue_test.dart

@@ -0,0 +1,50 @@
+import 'package:bluebubbles/services/network/attachment_download_queue.dart';
+import 'package:flutter_test/flutter_test.dart';
+
+void main() {
+  test("a tapped attachment moves ahead of older queued work in its chat", () {
+    final queue = AttachmentDownloadQueue<String>();
+    queue.add("chat", "first");
+    queue.add("chat", "second");
+    queue.add("chat", "third");
+
+    expect(queue.prioritize("chat", "third"), isTrue);
+    expect(queue.forChat("chat"), <String>["third", "first", "second"]);
+  });
+
+  test("prioritizing an already queued item does not duplicate it", () {
+    final queue = AttachmentDownloadQueue<String>();
+    queue.add("chat", "first");
+    queue.add("chat", "second");
+
+    queue.add("chat", "second", prioritized: true);
+
+    expect(queue.forChat("chat"), <String>["second", "first"]);
+  });
+
+  test("active chat ordering wins while preserving FIFO elsewhere", () {
+    final queue = AttachmentDownloadQueue<String>();
+    queue.add("other", "other-1");
+    queue.add("active", "active-1");
+    queue.add("active", "active-2");
+
+    expect(
+      queue.next(activeChatGuid: "active", isFetching: (_) => false),
+      "active-1",
+    );
+    queue.prioritize("active", "active-2");
+    expect(
+      queue.next(activeChatGuid: "active", isFetching: (_) => false),
+      "active-2",
+    );
+  });
+
+  test("prioritizing the first queued item preserves stable order", () {
+    final queue = AttachmentDownloadQueue<String>();
+    queue.add("chat", "active");
+    queue.add("chat", "waiting");
+
+    expect(queue.prioritize("chat", "active"), isTrue);
+    expect(queue.forChat("chat"), <String>["active", "waiting"]);
+  });
+}