Refactor local storage implementation in powersync_attachments_stream package. Updated attachment queue initialization to use localStorage, and enhanced README for clarity on storage handling. Added tests for edge cases and robustness in local storage operations.

Author: dean-journeyapps

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

@@ -1,5 +1,6 @@
 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_attachments_stream/powersync_attachments_stream.dart';
 import 'package:powersync_flutter_demo/attachments/camera_helpers.dart';
@@ -37,7 +38,8 @@ class _PhotoWidgetState extends State<PhotoWidget> {
     if (photoId == null) {
       return _ResolvedPhotoState(photoPath: null, fileExists: false);
     }
-    photoPath = await attachmentQueue.getLocalUri('$photoId.jpg');
+    final appDocDir = await getApplicationDocumentsDirectory();
+    photoPath = p.join(appDocDir.path, '$photoId.jpg');
 
     bool fileExists = await File(photoPath).exists();
 

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

@@ -19,7 +19,8 @@ Future<void> initializeAttachmentQueue(PowerSyncDatabase db) async {
     db: db,
     remoteStorage: remoteStorage,
     logger: log,
-    attachmentsDirectory: '${appDocDir.path}/attachments',
+    localStorage: IOLocalStorage(appDocDir.path),
+    // attachmentsDirectory: '${appDocDir.path}/attachments',
     watchAttachments: () => db.watch('''
       SELECT photo_id as id FROM todos WHERE photo_id IS NOT NULL
     ''').map((results) => results

packages/powersync_attachments_stream/README.md

@@ -44,11 +44,12 @@ Schema schema = Schema(([
 
 ```dart
 final Directory appDocDir = await getApplicationDocumentsDirectory();
+final localStorage = IOLocalStorage('${appDocDir.path}/attachments');
 
 final queue = AttachmentQueue(
     db: db,
     remoteStorage: remoteStorage,
-    attachmentsDirectory: '${appDocDir.path}/attachments',
+    localStorage: localStorage,
     watchAttachments: () => db.watch('''
       SELECT photo_id as id FROM todos WHERE photo_id IS NOT NULL
     ''').map((results) => results
@@ -60,7 +61,7 @@ final queue = AttachmentQueue(
   );
 ```
 
-* The `attachmentsDirectory` specifies where local attachment files should be stored. This directory needs to be provided to the constructor. In Flutter, `path_provider`'s `getApplicationDocumentsDirectory()` with a subdirectory like `/attachments` is a good choice.
+* The `localStorage` is an implementation of `AbstractLocalStorageAdapter` that specifies where and how local attachment files should be stored. For mobile and desktop apps, `IOLocalStorage` can be used, which requires a directory path. In Flutter, `path_provider`'s `getApplicationDocumentsDirectory()` with a subdirectory like `/attachments` is a good choice.
 * The `remoteStorage` is responsible for connecting to the attachments backend. See the `RemoteStorageAdapter` interface definition [here](https://github.com/powersync-ja/powersync.dart/blob/main/packages/powersync_attachments_stream/lib/src/abstractions/remote_storage.dart).
 * `watchAttachments` is a `Stream` of `WatchedAttachmentItem`. The `WatchedAttachmentItem`s represent the attachments which should be present in the application. We recommend using `PowerSync`'s `watch` query as shown above. In this example, we provide the `fileExtension` for all photos. This information could also be obtained from the query if necessary.
 
@@ -69,7 +70,7 @@ final queue = AttachmentQueue(
 ```dart
 final remote = _RemoteStorageAdapter();
 
-class _RemoteStorageAdapter implements RemoteStorageAdapter {
+class _RemoteStorageAdapter implements AbstractRemoteStorageAdapter {
   @override
   Future<void> uploadFile(Stream<List<int>> fileData, Attachment attachment) async {
     // TODO: Implement upload to your backend

packages/powersync_attachments_stream/analysis_options.yaml

@@ -1,4 +1,10 @@
 include: package:flutter_lints/flutter.yaml
 
+analyzer:
+  language:
+    strict-casts: true
+    strict-inference: true
+    strict-raw-types: true
+
 # Additional information about this file can be found at
 # https://dart.dev/guides/language/analysis-options

packages/powersync_attachments_stream/lib/src/abstractions/local_storage.dart

@@ -26,19 +26,9 @@ abstract class AbstractLocalStorageAdapter {
   /// Returns true if the file exists, false otherwise
   Future<bool> fileExists(String filePath);
 
-  /// Creates a directory at the specified path
-  ///
-  /// [path] - Path of the directory to create
-  Future<void> makeDir(String path);
-
-  /// Recursively removes a directory and its contents
-  ///
-  /// [path] - Path of the directory to remove
-  Future<void> rmDir(String path);
+  /// Initializes the storage, performing any necessary setup.
+  Future<void> initialize();
 
-  /// Copies a file from source to target path
-  ///
-  /// [sourcePath] - Path of the source file
-  /// [targetPath] - Path where the file will be copied
-  Future<void> copyFile(String sourcePath, String targetPath);
+  /// Clears all data from the storage.
+  Future<void> clear();
 }

packages/powersync_attachments_stream/lib/src/attachment_queue_service.dart

@@ -7,7 +7,6 @@
 import 'dart:async';
 import 'dart:typed_data';
 import 'package:logging/logging.dart';
-import './storage/io_local_storage.dart';
 import 'package:powersync_core/powersync_core.dart';
 import 'package:sqlite_async/mutex.dart';
 import 'attachment.dart';
@@ -75,7 +74,6 @@ class WatchedAttachmentItem {
 class AttachmentQueue {
   final PowerSyncDatabase db;
   final AbstractRemoteStorageAdapter remoteStorage;
-  final String attachmentsDirectory;
   final Stream<List<WatchedAttachmentItem>> Function() watchAttachments;
   final AbstractLocalStorageAdapter localStorage;
   final String attachmentsQueueTableName;
@@ -90,25 +88,23 @@ class AttachmentQueue {
 
   final Mutex _mutex = Mutex();
   bool _closed = false;
-  StreamSubscription? _syncStatusSubscription;
+  StreamSubscription<List<WatchedAttachmentItem>>? _syncStatusSubscription;
   late final AbstractAttachmentService attachmentsService;
   late final SyncingService syncingService;
 
   AttachmentQueue({
     required this.db,
     required this.remoteStorage,
-    required this.attachmentsDirectory,
     required this.watchAttachments,
-    IOLocalStorage? localStorage,
+    required this.localStorage,
     this.attachmentsQueueTableName = defaultTableName,
     this.errorHandler,
     this.syncInterval = const Duration(seconds: 30),
     this.archivedCacheLimit = 100,
     this.syncThrottleDuration = const Duration(seconds: 1),
     this.downloadAttachments = true,
     Logger? logger,
-  }) : localStorage = localStorage ?? IOLocalStorage(),
-       logger = logger ?? Logger('AttachmentQueue') {
+  }) : logger = logger ?? Logger('AttachmentQueue') {
     attachmentsService = AttachmentServiceImpl(
       db: db,
       logger: logger ?? Logger('AttachmentQueue'),
@@ -118,9 +114,8 @@ class AttachmentQueue {
 
     syncingService = SyncingService(
       remoteStorage: remoteStorage,
-      localStorage: this.localStorage,
+      localStorage: localStorage,
       attachmentsService: attachmentsService,
-      getLocalUri: (filename) async => getLocalUri(filename),
       errorHandler: errorHandler,
       syncThrottle: syncThrottleDuration,
       period: syncInterval,
@@ -139,7 +134,7 @@ class AttachmentQueue {
 
       await _stopSyncingInternal();
 
-      await localStorage.makeDir(attachmentsDirectory);
+      await localStorage.initialize();
 
       await attachmentsService.withContext((context) async {
         await _verifyAttachments(context);
@@ -256,7 +251,7 @@ class AttachmentQueue {
       }
 
       // Archive any items not specified in the watched items.
-      // For QUEUED_DELETE or QUEUED_UPLOAD states, archive only if hasSynced is true.
+      // For queuedDelete or queuedUpload states, archive only if hasSynced is true.
       // For other states, archive if the record is not found in the items.
       for (final attachment in currentAttachments) {
         final notInWatchedItems = items.every(
@@ -301,10 +296,9 @@ class AttachmentQueue {
       id,
       fileExtension,
     );
-    final String localUri = getLocalUri(filename);
 
     // Write the file to the filesystem.
-    final fileSize = await localStorage.saveFile(localUri, data);
+    final fileSize = await localStorage.saveFile(filename, data);
 
     return await attachmentsService.withContext((attachmentContext) async {
       return await db.writeTransaction((tx) async {
@@ -314,7 +308,7 @@ class AttachmentQueue {
           size: fileSize,
           mediaType: mediaType,
           state: AttachmentState.queuedUpload,
-          localUri: localUri,
+          localUri: filename,
           metaData: metaData,
         );
 
@@ -354,11 +348,6 @@ class AttachmentQueue {
     });
   }
 
-  /// Returns the user's storage directory with the attachment path used to load the file.
-  String getLocalUri(String filename) {
-    return '$attachmentsDirectory/$filename';
-  }
-
   /// Removes all archived items.
   Future<void> expireCache() async {
     await attachmentsService.withContext((context) async {
@@ -374,7 +363,7 @@ class AttachmentQueue {
     await attachmentsService.withContext((context) async {
       await context.clearQueue();
     });
-    await localStorage.rmDir(attachmentsDirectory);
+    await localStorage.clear();
   }
 
   /// Cleans up stale attachments.

packages/powersync_attachments_stream/lib/src/implementations/attachment_context.dart

@@ -124,8 +124,9 @@ class AttachmentContextImpl implements AbstractAttachmentContext {
     // Delete the archived attachments from the table
     final ids = archivedAttachments.map((a) => a.id).toList();
     if (ids.isNotEmpty) {
-      final placeholders = List.filled(ids.length, '?').join(',');
-      await db.execute('DELETE FROM $table WHERE id IN ($placeholders)', ids);
+      await db.executeBatch('DELETE FROM $table WHERE id = ?', [
+        for (final id in ids) [id],
+      ]);
     }
 
     log.info('Deleted ${archivedAttachments.length} archived attachments.');

packages/powersync_attachments_stream/lib/src/implementations/attachment_service.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
+import 'package:sqlite_async/sqlite_async.dart';
 
 import '../abstractions/attachment_service.dart';
 import '../abstractions/attachment_context.dart';
@@ -12,7 +13,7 @@ class AttachmentServiceImpl implements AbstractAttachmentService {
   final Logger logger;
   final int maxArchivedCount;
   final String attachmentsQueueTableName;
-  Future<void> _mutex = Future.value();
+  final Mutex _mutex = Mutex();
 
   late final AbstractAttachmentContext _context;
 
@@ -22,13 +23,18 @@ class AttachmentServiceImpl implements AbstractAttachmentService {
     required this.maxArchivedCount,
     required this.attachmentsQueueTableName,
   }) {
-    _context = AttachmentContextImpl(db, logger, maxArchivedCount, attachmentsQueueTableName);
+    _context = AttachmentContextImpl(
+      db,
+      logger,
+      maxArchivedCount,
+      attachmentsQueueTableName,
+    );
   }
 
   @override
   Stream<void> watchActiveAttachments() async* {
     logger.info('Watching attachments...');
-    
+
     // Watch for attachments with active states (queued for upload, download, or delete)
     final stream = db.watch(
       '''
@@ -54,10 +60,16 @@ class AttachmentServiceImpl implements AbstractAttachmentService {
   }
 
   @override
-  Future<T> withContext<T>(Future<T> Function(AbstractAttachmentContext ctx) action) {
-    // Simple mutex using chained futures
-    final completer = Completer<T>();
-    _mutex = _mutex.then((_) => action(_context)).then(completer.complete).catchError(completer.completeError);
-    return completer.future;
+  Future<T> withContext<T>(
+    Future<T> Function(AbstractAttachmentContext ctx) action,
+  ) async {
+    return await _mutex.lock(() async {
+      try {
+        return await action(_context);
+      } catch (e, stackTrace) {
+        // Re-throw the error to be handled by the caller
+        Error.throwWithStackTrace(e, stackTrace);
+      }
+    });
   }
-} 
+}

packages/powersync_attachments_stream/lib/src/storage/io_local_storage.dart

@@ -23,10 +23,12 @@ import '../abstractions/local_storage.dart';
 ///
 /// Handles file and directory operations for attachments.
 class IOLocalStorage implements AbstractLocalStorageAdapter {
-  final Directory baseDir;
+  final String attachmentsDirectory;
+  late final Directory baseDir;
 
-  IOLocalStorage([Directory? baseDir])
-    : baseDir = baseDir ?? Directory.systemTemp;
+  IOLocalStorage(this.attachmentsDirectory) {
+    baseDir = Directory(attachmentsDirectory);
+  }
 
   File _fileFor(String filePath) => File(p.join(baseDir.path, filePath));
   File _metaFileFor(String filePath) =>
@@ -88,40 +90,15 @@ class IOLocalStorage implements AbstractLocalStorageAdapter {
 
   /// Creates a directory and all necessary parent directories dynamically if they do not exist.
   @override
-  Future<void> makeDir(String path) async {
-    await Directory(p.join(baseDir.path, path)).create(recursive: true);
+  Future<void> initialize() async {
+    await baseDir.create(recursive: true);
   }
 
-  /// Recursively removes a directory and all its contents.
   @override
-  Future<void> rmDir(String path) async {
-    final dir = Directory(p.join(baseDir.path, path));
-    if (await dir.exists()) {
-      await for (final entity in dir.list(recursive: false)) {
-        if (entity is Directory) {
-          await rmDir(p.relative(entity.path, from: baseDir.path));
-        } else if (entity is File) {
-          await entity.delete();
-        }
-      }
-      await dir.delete();
-    }
-  }
-
-  /// Copies a file and its metadata to a new location.
-  @override
-  Future<void> copyFile(String sourcePath, String targetPath) async {
-    final sourceFile = _fileFor(sourcePath);
-    final targetFile = _fileFor(targetPath);
-    if (!await sourceFile.exists()) {
-      throw FileSystemException('Source file does not exist', sourcePath);
-    }
-    await targetFile.parent.create(recursive: true);
-    await sourceFile.copy(targetFile.path);
-    final sourceMeta = _metaFileFor(sourcePath);
-    final targetMeta = _metaFileFor(targetPath);
-    if (await sourceMeta.exists()) {
-      await sourceMeta.copy(targetMeta.path);
+  Future<void> clear() async {
+    if (await baseDir.exists()) {
+      await baseDir.delete(recursive: true);
     }
+    await baseDir.create(recursive: true);
   }
 }