Also listen for connection status

Author: simolus3

packages/powersync_core/doc/attachments.md

@@ -91,7 +91,7 @@ Next, start the sync process by calling `attachmentQueue.startSync()`.
 
 To create a new attachment locally, call `AttachmentQueue.saveFile`. To represent the attachment, this method takes
 the contents to store, the media type, an optional file extension and id.
-The queue will store the contents in a local file and mark is as queued for uploads. It also invokes a callback
+The queue will store the contents in a local file and mark it as queued for upload. It also invokes a callback
 responsible for referencing the id of the generated attachment in the primary data model:
 
 ```dart

packages/powersync_core/lib/src/attachments/attachment_queue_service.dart

@@ -73,7 +73,8 @@ base class AttachmentQueue {
 
   final Mutex _mutex = Mutex();
   bool _closed = false;
-  StreamSubscription<List<WatchedAttachmentItem>>? _syncStatusSubscription;
+  StreamSubscription<void>? _syncStatusSubscription;
+  StreamSubscription<void>? _watchedAttachmentsSubscription;
   final AttachmentService _attachmentsService;
   final SyncingService _syncingService;
 
@@ -168,10 +169,23 @@ base class AttachmentQueue {
         await _verifyAttachments(context);
       });
 
+      // Listen for connectivity changes and watched attachments
       await _syncingService.startSync();
 
-      // Listen for connectivity changes and watched attachments
-      _syncStatusSubscription = _watchAttachments().listen((items) async {
+      _watchedAttachmentsSubscription =
+          _watchAttachments().listen((items) async {
+        await _processWatchedAttachments(items);
+      });
+
+      var previouslyConnected = _db.currentStatus.connected;
+      _syncStatusSubscription = _db.statusStream.listen((status) {
+        if (!previouslyConnected && status.connected) {
+          _syncingService.triggerSync();
+        }
+
+        previouslyConnected = status.connected;
+      });
+      _watchAttachments().listen((items) async {
         await _processWatchedAttachments(items);
       });
 
@@ -187,10 +201,19 @@ base class AttachmentQueue {
   }
 
   Future<void> _stopSyncingInternal() async {
-    if (_closed || _syncStatusSubscription == null) return;
+    if (_closed ||
+        _syncStatusSubscription == null ||
+        _watchedAttachmentsSubscription == null) {
+      return;
+    }
+
+    await (
+      _syncStatusSubscription!.cancel(),
+      _watchedAttachmentsSubscription!.cancel(),
+    ).wait;
 
-    await _syncStatusSubscription?.cancel();
     _syncStatusSubscription = null;
+    _watchedAttachmentsSubscription = null;
     await _syncingService.stopSync();
 
     _logger.info('AttachmentQueue stopped syncing.');

packages/powersync_core/lib/src/attachments/sync/syncing_service.dart

@@ -85,9 +85,8 @@ final class SyncingService {
   }
 
   /// Enqueues a sync operation (manual trigger).
-  Future<void> triggerSync() async {
-    if (_isClosed) return;
-    _syncTriggerController.add(null);
+  void triggerSync() {
+    if (!_isClosed) _syncTriggerController.add(null);
   }
 
   /// Stops all ongoing sync operations.