Refactor attachment handling and logging in the powersync_attachments_stream package. Removed unused imports, improved error handling during file uploads and downloads, and enhanced and renamed logging for better traceability of attachment operations.

Author: dean-journeyapps

demos/supabase-todolist-new-attachment/lib/attachments/photo_capture_widget.dart

@@ -4,10 +4,7 @@ import 'dart:typed_data';
 import 'package:camera/camera.dart';
 import 'package:flutter/material.dart';
 import 'package:logging/logging.dart';
-import 'package:powersync/powersync.dart' as powersync;
 import 'package:powersync_flutter_demo_new/attachments/queue.dart';
-import 'package:powersync_flutter_demo_new/models/todo_item.dart';
-import 'package:powersync_flutter_demo_new/powersync.dart';
 
 class TakePhotoWidget extends StatefulWidget {
   final String todoId;

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

@@ -38,9 +38,8 @@ Future<void> initializeAttachmentQueue(PowerSyncDatabase db) async {
       return items;
     }),
     localStorage: localStorage,
-    errorHandler: null, // Optionally implement SyncErrorHandler
+    errorHandler: null, 
   );
-  // await attachmentQueue.startSync();
 }
 
 Future<Attachment> savePhotoAttachment(Stream<Uint8List> photoData, String todoId,

demos/supabase-todolist-new-attachment/lib/attachments/remote_storage_adapter.dart

@@ -3,30 +3,57 @@ import 'dart:typed_data';
 import 'package:powersync_attachments_stream/powersync_attachments_stream.dart';
 import 'package:powersync_flutter_demo_new/app_config.dart';
 import 'package:supabase_flutter/supabase_flutter.dart';
-import 'package:image/image.dart' as img;
+import 'package:logging/logging.dart';
 
 class SupabaseStorageAdapter implements RemoteStorage {
+  static final _log = Logger('SupabaseStorageAdapter');
+
   @override
   Future<void> uploadFile(
       Stream<List<int>> fileData, Attachment attachment) async {
     _checkSupabaseBucketIsConfigured();
-    final tempFile =
-        File('${Directory.systemTemp.path}/${attachment.filename}');
-    final sink = tempFile.openWrite();
+
+    // Check if attachment size is specified (required for buffer allocation)
+    final byteSize = attachment.size;
+    if (byteSize == null) {
+      throw Exception('Cannot upload a file with no byte size specified');
+    }
+
+    _log.info('uploadFile: ${attachment.filename} (size: $byteSize bytes)');
+
+    // Collect all stream data into a single Uint8List buffer (like Kotlin version)
+    final buffer = Uint8List(byteSize);
+    var position = 0;
+
     await for (final chunk in fileData) {
-      sink.add(chunk);
+      if (position + chunk.length > byteSize) {
+        throw Exception('File data exceeds specified size');
+      }
+      buffer.setRange(position, position + chunk.length, chunk);
+      position += chunk.length;
     }
-    await sink.close();
-    print('uploadFile: ${attachment.filename}');
+
+    if (position != byteSize) {
+      throw Exception(
+          'File data size ($position) does not match specified size ($byteSize)');
+    }
+
+    // Create a temporary file from the buffer for upload
+    final tempFile =
+        File('${Directory.systemTemp.path}/${attachment.filename}');
     try {
+      await tempFile.writeAsBytes(buffer);
+
       await Supabase.instance.client.storage
           .from(AppConfig.supabaseStorageBucket)
           .upload(attachment.filename, tempFile,
               fileOptions: FileOptions(
                   contentType:
                       attachment.mediaType ?? 'application/octet-stream'));
 
+      _log.info('Successfully uploaded ${attachment.filename}');
     } catch (error) {
+      _log.severe('Error uploading ${attachment.filename}', error);
       throw Exception(error);
     } finally {
       if (await tempFile.exists()) {
@@ -38,23 +65,26 @@ class SupabaseStorageAdapter implements RemoteStorage {
   @override
   Future<Stream<List<int>>> downloadFile(Attachment attachment) async {
     _checkSupabaseBucketIsConfigured();
-    print('downloadFile: ${attachment.filename}');
     try {
+      _log.info('downloadFile: ${attachment.filename}');
+
       Uint8List fileBlob = await Supabase.instance.client.storage
           .from(AppConfig.supabaseStorageBucket)
           .download(attachment.filename);
-      final image = img.decodeImage(fileBlob);
-      Uint8List blob = img.JpegEncoder().encode(image!);
-      print('downloadFile: ${blob.length}');
-      return Stream.value(blob);
+
+      _log.info(
+          'Successfully downloaded ${attachment.filename} (${fileBlob.length} bytes)');
+
+      // Return the raw file data as a stream
+      return Stream.value(fileBlob);
     } catch (error) {
+      _log.severe('Error downloading ${attachment.filename}', error);
       throw Exception(error);
     }
   }
 
   @override
   Future<void> deleteFile(Attachment attachment) async {
-    print('deleteFile: ${attachment.filename}');
     _checkSupabaseBucketIsConfigured();
     try {
       await Supabase.instance.client.storage

packages/powersync_attachments_stream/lib/src/attachment_queue_service.dart

@@ -71,7 +71,7 @@ class AttachmentQueue {
   }) : logger = logger ?? Logger('AttachmentQueue') {
     attachmentsService = AttachmentServiceImpl(
       db: db,
-      log: logger ?? Logger('AttachmentQueue'),
+      logger: logger ?? Logger('AttachmentQueue'),
       maxArchivedCount: archivedCacheLimit,
       attachmentsQueueTableName: attachmentsQueueTableName,
     );
@@ -195,12 +195,6 @@ class AttachmentQueue {
             ),
           );
 
-          // attachmentUpdates.add(
-          //   existingQueueItem.copyWith(
-          //     filename: filename,
-          //     state: AttachmentState.queuedDownload,
-          //   ),
-          // );
         } else if (existingQueueItem.state == AttachmentState.archived) {
           // The attachment is present again. Need to queue it for sync.
           if (existingQueueItem.hasSynced) {

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

@@ -101,30 +101,35 @@ class AttachmentContextImpl implements AttachmentContext {
   Future<bool> deleteArchivedAttachments(
     Future<void> Function(List<Attachment>) callback,
   ) async {
-    // Find all archived attachments
-    final results = await db.getAll('SELECT * FROM $table WHERE state = ?', [
-      AttachmentState.archived.index,
-    ]);
-    final archivedAttachments = results
-        .map((row) => Attachment.fromRow(row))
-        .toList();
+    // 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 ?',
+      [
+        AttachmentState.archived.index,
+        limit,
+        maxArchivedCount,
+      ],
+    );
+    final archivedAttachments = results.map((row) => Attachment.fromRow(row)).toList();
 
     if (archivedAttachments.isEmpty) {
       return false;
     }
 
-    log.info('Deleting ${archivedAttachments.length} archived attachments...');
+    log.info('Deleting ${archivedAttachments.length} archived attachments (exceeding maxArchivedCount=$maxArchivedCount)...');
     // Call the callback with the list of archived attachments before deletion
     await callback(archivedAttachments);
 
     // Delete the archived attachments from the table
     final ids = archivedAttachments.map((a) => a.id).toList();
-    // Use a batch delete for efficiency
-    final placeholders = List.filled(ids.length, '?').join(',');
-    await db.execute('DELETE FROM $table WHERE id IN ($placeholders)', ids);
+    if (ids.isNotEmpty) {
+      final placeholders = List.filled(ids.length, '?').join(',');
+      await db.execute('DELETE FROM $table WHERE id IN ($placeholders)', ids);
+    }
 
     log.info('Deleted ${archivedAttachments.length} archived attachments.');
-    return true;
+    return archivedAttachments.length < limit;
   }
 
   @override

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

@@ -9,7 +9,7 @@ import 'attachment_context.dart';
 
 class AttachmentServiceImpl implements AttachmentService {
   final PowerSyncDatabase db;
-  final Logger log;
+  final Logger logger;
   final int maxArchivedCount;
   final String attachmentsQueueTableName;
   Future<void> _mutex = Future.value();
@@ -18,16 +18,16 @@ class AttachmentServiceImpl implements AttachmentService {
 
   AttachmentServiceImpl({
     required this.db,
-    required this.log,
+    required this.logger,
     required this.maxArchivedCount,
     required this.attachmentsQueueTableName,
   }) {
-    _context = AttachmentContextImpl(db, log, maxArchivedCount, attachmentsQueueTableName);
+    _context = AttachmentContextImpl(db, logger, maxArchivedCount, attachmentsQueueTableName);
   }
 
   @override
   Stream<void> watchActiveAttachments() async* {
-    log.info('Watching attachments...');
+    logger.info('Watching attachments...');
 
     // Watch for attachments with active states (queued for upload, download, or delete)
     final stream = db.watch(