Unit tests for attachments

Author: simolus3

demos/supabase-todolist/lib/attachments/photo_widget.dart

@@ -3,7 +3,7 @@ import 'dart:io';
 import 'package:path_provider/path_provider.dart';
 import 'package:path/path.dart' as p;
 import 'package:flutter/material.dart';
-import 'package:powersync_core/attachments.dart';
+import 'package:powersync_core/attachments/attachments.dart';
 import 'package:powersync_flutter_demo/attachments/camera_helpers.dart';
 import 'package:powersync_flutter_demo/attachments/photo_capture_widget.dart';
 

packages/powersync_core/lib/attachments/attachments.dart

@@ -9,3 +9,4 @@ export '../src/attachments/attachment.dart';
 export '../src/attachments/attachment_queue_service.dart';
 export '../src/attachments/local_storage.dart';
 export '../src/attachments/remote_storage.dart';
+export '../src/attachments/sync_error_handler.dart';

packages/powersync_core/lib/attachments/io.dart

@@ -1,7 +1,7 @@
 /// A platform-specific import supporting attachments on native platforms.
 ///
 /// This library exports the [IOLocalStorage] class, implementing the
-/// [LocalStorageAdapter] interface by storing files under a root directory.
+/// [LocalStorage] interface by storing files under a root directory.
 ///
 /// {@category attachments}
 library;

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

@@ -142,6 +142,11 @@ final class Attachment {
       metaData: metaData ?? this.metaData,
     );
   }
+
+  @override
+  String toString() {
+    return 'Attachment(id: $id, state: $state, localUri: $localUri, metadata: $metaData)';
+  }
 }
 
 /// Table definition for the attachments queue.

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

@@ -46,7 +46,7 @@ final class WatchedAttachmentItem {
   /// Creates a [WatchedAttachmentItem].
   ///
   /// Either [fileExtension] or [filename] must be provided.
-  WatchedAttachmentItem({
+  const WatchedAttachmentItem({
     required this.id,
     this.fileExtension,
     this.filename,
@@ -67,7 +67,7 @@ final class WatchedAttachmentItem {
 base class AttachmentQueue {
   final PowerSyncDatabase _db;
   final Stream<List<WatchedAttachmentItem>> Function() _watchAttachments;
-  final LocalStorageAdapter _localStorage;
+  final LocalStorage _localStorage;
   final bool _downloadAttachments;
   final Logger _logger;
 
@@ -80,7 +80,7 @@ base class AttachmentQueue {
   AttachmentQueue._(
       {required PowerSyncDatabase db,
       required Stream<List<WatchedAttachmentItem>> Function() watchAttachments,
-      required LocalStorageAdapter localStorage,
+      required LocalStorage localStorage,
       required bool downloadAttachments,
       required Logger logger,
       required AttachmentService attachmentsService,
@@ -110,9 +110,9 @@ base class AttachmentQueue {
   /// - [logger]: Logging interface used for all log operations.
   factory AttachmentQueue({
     required PowerSyncDatabase db,
-    required RemoteStorageAdapter remoteStorage,
+    required RemoteStorage remoteStorage,
     required Stream<List<WatchedAttachmentItem>> Function() watchAttachments,
-    required LocalStorageAdapter localStorage,
+    required LocalStorage localStorage,
     String attachmentsQueueTableName = AttachmentsQueueTable.defaultTableName,
     AttachmentErrorHandler? errorHandler,
     Duration syncInterval = const Duration(seconds: 30),
@@ -136,6 +136,7 @@ base class AttachmentQueue {
       errorHandler: errorHandler,
       syncThrottle: syncThrottleDuration,
       period: syncInterval,
+      logger: resolvedLogger,
     );
 
     return AttachmentQueue._(
@@ -156,7 +157,7 @@ base class AttachmentQueue {
   Future<void> startSync() async {
     await _mutex.lock(() async {
       if (_closed) {
-        throw Exception('Attachment queue has been closed');
+        throw StateError('Attachment queue has been closed');
       }
 
       await _stopSyncingInternal();
@@ -186,7 +187,7 @@ base class AttachmentQueue {
   }
 
   Future<void> _stopSyncingInternal() async {
-    if (_closed) return;
+    if (_closed || _syncStatusSubscription == null) return;
 
     await _syncStatusSubscription?.cancel();
     _syncStatusSubscription = null;
@@ -200,10 +201,8 @@ base class AttachmentQueue {
     await _mutex.lock(() async {
       if (_closed) return;
 
-      await _syncStatusSubscription?.cancel();
-      await _syncingService.close();
+      await _stopSyncingInternal();
       _closed = true;
-
       _logger.info('AttachmentQueue closed.');
     });
   }
@@ -319,7 +318,7 @@ base class AttachmentQueue {
   }) async {
     final resolvedId = id ?? await generateAttachmentId();
 
-    final String filename = await resolveNewAttachmentFilename(
+    final filename = await resolveNewAttachmentFilename(
       resolvedId,
       fileExtension,
     );

packages/powersync_core/lib/src/attachments/implementations/attachment_context.dart

@@ -55,7 +55,7 @@ final class AttachmentContext {
 
   Future<void> saveAttachments(List<Attachment> attachments) async {
     if (attachments.isEmpty) {
-      log.info('No attachments to save.');
+      log.finer('No attachments to save.');
       return;
     }
     await db.writeTransaction((tx) async {
@@ -98,6 +98,7 @@ final class AttachmentContext {
   ) async {
     // Only delete archived attachments exceeding the maxArchivedCount, ordered by timestamp DESC
     const limit = 1000;
+
     final results = await db.getAll(
       'SELECT * FROM $table WHERE state = ? ORDER BY timestamp DESC LIMIT ? OFFSET ?',
       [
@@ -134,6 +135,8 @@ final class AttachmentContext {
     Attachment attachment,
     SqliteWriteContext context,
   ) async {
+    log.finest('Updating attachment ${attachment.id}: ${attachment.state}');
+
     await context.execute(
       '''INSERT OR REPLACE INTO 
                     $table (id, timestamp, filename, local_uri, media_type, size, state, has_synced, meta_data) 

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

@@ -7,13 +7,13 @@ import 'package:path/path.dart' as p;
 
 import 'local_storage.dart';
 
-/// Implements [LocalStorageAdapter] for device filesystem using Dart IO.
+/// Implements [LocalStorage] for device filesystem using Dart IO.
 ///
 /// Handles file and directory operations for attachments.
 ///
 /// {@category attachments}
 @experimental
-final class IOLocalStorage implements LocalStorageAdapter {
+final class IOLocalStorage implements LocalStorage {
   final Directory _root;
 
   const IOLocalStorage(this._root);

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

@@ -4,6 +4,7 @@ library;
 import 'dart:typed_data';
 
 import 'package:meta/meta.dart';
+import 'package:path/path.dart' as p;
 
 /// An interface responsible for storing attachment data locally.
 ///
@@ -15,7 +16,10 @@ import 'package:meta/meta.dart';
 ///
 /// {@category attachments}
 @experimental
-abstract interface class LocalStorageAdapter {
+abstract interface class LocalStorage {
+  /// Returns an in-memory [LocalStorage] implementation, suitable for testing.
+  factory LocalStorage.inMemory() = _InMemoryStorage;
+
   /// Saves binary data stream to storage at the specified file path
   ///
   /// [filePath] - Path where the file will be stored
@@ -48,3 +52,51 @@ abstract interface class LocalStorageAdapter {
   /// Clears all data from the storage.
   Future<void> clear();
 }
+
+final class _InMemoryStorage implements LocalStorage {
+  final Map<String, Uint8List> content = {};
+
+  String _keyForPath(String path) {
+    return p.normalize(path);
+  }
+
+  @override
+  Future<void> clear() async {
+    content.clear();
+  }
+
+  @override
+  Future<void> deleteFile(String filePath) async {
+    content.remove(_keyForPath(filePath));
+  }
+
+  @override
+  Future<bool> fileExists(String filePath) async {
+    return content.containsKey(_keyForPath(filePath));
+  }
+
+  @override
+  Future<void> initialize() async {}
+
+  @override
+  Stream<Uint8List> readFile(String filePath, {String? mediaType}) {
+    return switch (content[_keyForPath(filePath)]) {
+      null =>
+        Stream.error('file at $filePath does not exist in in-memory storage'),
+      final contents => Stream.value(contents),
+    };
+  }
+
+  @override
+  Future<int> saveFile(String filePath, Stream<List<int>> data) async {
+    var length = 0;
+    final builder = BytesBuilder(copy: false);
+    await for (final chunk in data) {
+      length += chunk.length;
+      builder.add(chunk);
+    }
+
+    content[_keyForPath(filePath)] = builder.takeBytes();
+    return length;
+  }
+}

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

@@ -10,7 +10,7 @@ import 'attachment.dart';
 ///
 /// {@category attachments}
 @experimental
-abstract interface class RemoteStorageAdapter {
+abstract interface class RemoteStorage {
   /// Uploads a file to remote storage.
   ///
   /// [fileData] is a stream of byte arrays representing the file data.

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

@@ -24,8 +24,8 @@ import '../sync_error_handler.dart';
 /// - [errorHandler]: Optional error handler for managing sync-related errors.
 @internal
 final class SyncingService {
-  final RemoteStorageAdapter remoteStorage;
-  final LocalStorageAdapter localStorage;
+  final RemoteStorage remoteStorage;
+  final LocalStorage localStorage;
   final AttachmentService attachmentsService;
   final AttachmentErrorHandler? errorHandler;
   final Duration syncThrottle;
@@ -44,8 +44,8 @@ final class SyncingService {
     this.errorHandler,
     this.syncThrottle = const Duration(seconds: 5),
     this.period = const Duration(seconds: 30),
-    Logger? logger,
-  }) : logger = logger ?? Logger('SyncingService');
+    required this.logger,
+  });
 
   /// Starts the syncing process, including periodic and event-driven sync operations.
   Future<void> startSync() async {
@@ -60,11 +60,11 @@ final class SyncingService {
     );
     final manualTriggers = _syncTriggerController.stream;
 
-    // Merge both streams and apply throttling
-    _syncSubscription =
-        StreamGroup.merge<void>([attachmentChanges, manualTriggers]).listen((
-      _,
-    ) async {
+    late StreamSubscription<void> sub;
+    final syncStream =
+        StreamGroup.merge<void>([attachmentChanges, manualTriggers])
+            .takeWhile((_) => sub == _syncSubscription)
+            .asyncMap((_) async {
       await attachmentsService.withContext((context) async {
         final attachments = await context.getActiveAttachments();
         logger.info('Found ${attachments.length} active attachments');
@@ -73,6 +73,8 @@ final class SyncingService {
       });
     });
 
+    _syncSubscription = sub = syncStream.listen(null);
+
     // Start periodic sync using instance period
     _periodicSubscription = Stream<void>.periodic(period, (_) {}).listen((
       _,
@@ -90,8 +92,16 @@ final class SyncingService {
 
   /// Stops all ongoing sync operations.
   Future<void> stopSync() async {
-    await _syncSubscription?.cancel();
     await _periodicSubscription?.cancel();
+
+    final subscription = _syncSubscription;
+    // Add a trigger event after clearing the subscription, which will make
+    // the takeWhile() callback cancel. This allows us to use asFuture() here,
+    // ensuring that we only complete this future when the stream is actually
+    // done.
+    _syncSubscription = null;
+    _syncTriggerController.add(null);
+    await subscription?.asFuture<void>();
   }
 
   /// Closes the syncing service, stopping all operations and releasing resources.
@@ -177,7 +187,8 @@ final class SyncingService {
         st,
       );
       if (errorHandler != null) {
-        final shouldRetry = await errorHandler!.onUploadError(attachment, e);
+        final shouldRetry =
+            await errorHandler!.onUploadError(attachment, e, st);
         if (!shouldRetry) {
           logger.info('Attachment with ID ${attachment.id} has been archived');
           return attachment.copyWith(state: AttachmentState.archived);
@@ -206,7 +217,8 @@ final class SyncingService {
       );
     } catch (e, st) {
       if (errorHandler != null) {
-        final shouldRetry = await errorHandler!.onDownloadError(attachment, e);
+        final shouldRetry =
+            await errorHandler!.onDownloadError(attachment, e, st);
         if (!shouldRetry) {
           logger.info('Attachment with ID ${attachment.id} has been archived');
           return attachment.copyWith(state: AttachmentState.archived);
@@ -240,7 +252,8 @@ final class SyncingService {
       return attachment.copyWith(state: AttachmentState.archived);
     } catch (e, st) {
       if (errorHandler != null) {
-        final shouldRetry = await errorHandler!.onDeleteError(attachment, e);
+        final shouldRetry =
+            await errorHandler!.onDeleteError(attachment, e, st);
         if (!shouldRetry) {
           logger.info('Attachment with ID ${attachment.id} has been archived');
           return attachment.copyWith(state: AttachmentState.archived);