Attachments orphan issue on web (powersync_core)

[galacticgibbon]

I have a scenario on web where a failure occurs - specifically with web using LocalStorage.inMemory()

Reproduce:

1. Add an item to the queue
2. Fail the upload (for example offline)
3. Refresh the page.

The data will be gone from the local storage, no more attempts are made to upload, but the attachment still exists (and any relations added in watchAttachments).

What I expect

I think there needs to be some way to detect these failures reliably so that you can delete any associated entries.

[Chriztiaan]

Hey @galacticgibbon, inMemory is great for testing - but generally I'd recommend that you use a different type of of storage to be able to handle this scenario. Could you share your implementation?

[galacticgibbon]

Yep, you're probably right. That being said, not all websites can support OPFS (mine cannot), and IndexedDB is not sufficient with its memory limit.

In the meantime, I decided to bypass the attachment queue on web, and just use a direct upload.

Here's some snippets of my setup:

    for (final image in imagesToUpload) {
      // Compress first
      final (:bytes, :wasCompressed) = await compressionService.compressImage(
        image,
      );

      final strippedFilename = replaceExtension(image.name, '');

      String mimeType;
      final String? fileExtension;

      if (wasCompressed) {
        fileExtension = 'jpg';
        mimeType = 'image/jpeg';
      } else {
        final lookupMime = lookupMimeType(
          image.name,
          headerBytes: bytes.toList(),
        );

        if (lookupMime == null) {
          continue;
        }

        fileExtension = getFileExtension(image.name);

        mimeType = lookupMime;
      }

      if (!['image/jpeg', 'image/png'].contains(mimeType)) {
        continue;
      }

      // Generate unique filename for storage
      final filename =
          '$strippedFilename${fileExtension?.let((ext) => '.$ext') ?? ''}';

      // Create metadata for attachment queue/loader
      final metadata = AttachmentMetadata(entryId: entryId);

      // Save file to queue (Native) or Direct Upload (Web)
      await attachmentLoader.saveFile(
        data: Stream.value(bytes),
        mediaType: mimeType,
        metaData: metadata.toJson(),
        updateHook: (tx, attachmentId) async {
          await repository.insertAttachment(
            tx: tx,
            id: attachmentId,
            firestoreId: attachmentId,
            entryId: entryId,
            filePath: _generateFilePath(entryId, attachmentId),
            filename: filename,
            size: bytes.length,
            kind: AttachmentKind.image.name,
            mimeType: mimeType,
          );
        },
      );
    }

class LocalAttachmentSource extends AttachmentSource {
  final String path;
  const LocalAttachmentSource(this.path);
}

class NetworkAttachmentSource extends AttachmentSource {
  final String url;
  const NetworkAttachmentSource(this.url);
}

abstract class AttachmentLoader {
  Future<void> saveFile({
    required Stream<List<int>> data,
    required String mediaType,
    required String metaData,
    required Future<void> Function(SqliteWriteContext tx, String attachmentId)
    updateHook,
  });

  Future<Map<String, AttachmentSource>> getAttachmentSources(
    List<EntryAttachment> attachments,
  );

  Future<void> deleteAttachment({
    required EntryAttachment attachment,
    required Future<void> Function(SqliteWriteContext tx, String attachmentId)
    updateHook,
  });
}

class AttachmentUploaderNative implements AttachmentLoader {
  final AttachmentQueue _queue;

  AttachmentUploaderNative(this._queue);

  @override
  Future<void> saveFile({
    required Stream<List<int>> data,
    required String mediaType,
    required String metaData,
    required Future<void> Function(SqliteWriteContext tx, String attachmentId)
    updateHook,
  }) async {
    await _queue.saveFile(
      data: data,
      mediaType: mediaType,
      metaData: metaData,
      updateHook: (tx, attachment) async {
        await updateHook(tx, attachment.id);
      },
    );
  }

  @override
  Future<Map<String, AttachmentSource>> getAttachmentSources(
    List<EntryAttachment> attachments,
  ) async {
    final basePath = await getAttachmentBasePath('entry_attachments');

    return {
      for (final attachment in attachments)
        attachment.id: LocalAttachmentSource('$basePath/${attachment.id}.dat'),
    };
  }

  @override
  Future<void> deleteAttachment({
    required EntryAttachment attachment,
    required Future<void> Function(SqliteWriteContext tx, String attachmentId)
    updateHook,
  }) async {
    await _queue.deleteFile(
      attachmentId: attachment.id,
      updateHook: (tx, attachment) async {
        await updateHook(tx, attachment.id);
      },
    );
  }
}

import 'dart:typed_data';

import 'package:bullet_common/src/features/attachments/domain/entry_attachment.dart';
import 'package:dio/dio.dart';
import 'package:powersync_core/powersync_core.dart';
import 'package:sqlite_async/sqlite_async.dart';

import '../attachment_api_client.dart';
import '../attachment_metadata.dart';
import 'attachment_loader.dart';

class AttachmentUploaderWeb implements AttachmentLoader {
  final AttachmentApiClient _apiClient;
  final PowerSyncDatabase _db;
  final Dio _dio;

  AttachmentUploaderWeb(this._apiClient, this._db) : _dio = Dio();

  @override
  Future<void> saveFile({
    required Stream<List<int>> data,
    required String mediaType,
    required String metaData,
    required Future<void> Function(SqliteWriteContext tx, String attachmentId)
    updateHook,
  }) async {
    final attachmentId = uuid.v4();

    // 1. Parse metadata to get entryId and filename
    final metadata = AttachmentMetadata.fromJson(metaData);

    // 2. Get signed upload URL
    final uploadUrl = await _apiClient.getUploadUrl((
      entryId: metadata.entryId,
      attachmentId: attachmentId,
    ));

    // 3. Upload file directly
    final bytes = await _streamToBytes(data);

    // 4. Upload file to signed URL
    await _dio.put(
      uploadUrl,
      data: Stream.fromIterable([bytes]),
      options: Options(
        headers: {'Content-Type': mediaType, 'Content-Length': bytes.length},
      ),
    );

    await _db.writeTransaction((tx) async {
      await updateHook(tx, attachmentId);
    });
  }

  @override
  Future<Map<String, AttachmentSource>> getAttachmentSources(
    List<EntryAttachment> attachments,
  ) async {
    if (attachments.isEmpty) return {};

    final urls = await _apiClient.getBatchDownloadUrls(
      attachments: attachments
          .map((e) => (entryId: e.entryId, attachmentId: e.id))
          .toList(),
    );

    return {
      for (final attachment in attachments)
        // Build composite key to lookup URL
        if (urls['${attachment.entryId}/${attachment.id}'] case String url)
          attachment.id: NetworkAttachmentSource(url),
    };
  }

  @override
  Future<void> deleteAttachment({
    required EntryAttachment attachment,
    required Future<void> Function(SqliteWriteContext tx, String attachmentId)
    updateHook,
  }) async {
    await _apiClient.deleteFile(
      params: (entryId: attachment.entryId, attachmentId: attachment.id),
    );
    await _db.writeTransaction((tx) async {
      await updateHook(tx, attachment.id);
    });
  }

  Future<Uint8List> _streamToBytes(Stream<List<int>> stream) async {
    final builder = BytesBuilder(copy: false);
    await for (final chunk in stream) {
      builder.add(chunk);
    }
    return builder.takeBytes();
  }
}

import 'package:bullet_common/src/infra/attachments/attachment_api_client.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:powersync_core/attachments/attachments.dart';
import 'package:riverpod_annotation/riverpod_annotation.dart';

import '../../../database/powersync.dart';
import '../attachment_metadata.dart';
import 'attachment_remote_storage.dart';
import 'attachment_storage_js.dart'
    if (dart.library.io) 'attachment_storage_native.dart';

part 'attachment_queue.g.dart';

@riverpod
Future<AttachmentQueue> attachmentQueue(Ref ref) async {
  final powersync = await ref.watch(powerSyncDatabaseProvider.future);

  final queue = AttachmentQueue(
    db: powersync,
    localStorage: await attachmentStorage('entry_attachments'),
    remoteStorage: AttachmentRemoteStorage(
      apiClient: ref.watch(attachmentApiClientProvider),
    ),
    watchAttachments: () {
      // Watch entry_attachments table for new attachments
      return powersync
          .watch('SELECT id, filename, entry_id FROM entry_attachments')
          .map(
            (results) => results.map((row) {
              final entryId = row['entry_id'] as String;
              final filename = row['filename'] as String;

              // Build metadata JSON
              final metadata = AttachmentMetadata(entryId: entryId);

              return WatchedAttachmentItem(
                id: row['id'] as String,
                filename: filename,
                metaData: metadata.toJson(),
              );
            }).toList(),
          );
    },
  );

  await queue.startSync();

  ref.onDispose(() {
    queue.stopSyncing();
  });

  return queue;
}

[galacticgibbon]

Since this is experimental, I had some confusion around file names, file extensions etc. It would be great if the documentation had some information about how files are saved on the client etc. It wasn't super clear how WatchedAttachmentItem worked, but altogether very happy with it!