Attachments / Files

# Built into @powersync/web, @powersync/node, and @powersync/react-native.
# React Native requires an additional storage adapter:
# npm install @powersync/attachments-storage-react-native

# Expo projects (v54+) also require:
# npm install expo-file-system

# Bare React Native projects also require:
# npm install @dr.pogodin/react-native-fs

# Built into the powersync package (v1.16.0+). No additional installation needed.

// Built into the PowerSync Swift SDK (v1.0.0+). No additional installation needed.

// Built into the PowerSync Kotlin SDK (v1.0.0+). No additional installation needed.

# Built into PowerSync.Common (v0.1.2+). No additional installation needed.

import { Schema, Table, column, AttachmentTable } from '@powersync/web';

const appSchema = new Schema({
  users: new Table({
    name: column.text,
    email: column.text,
    photo_id: column.text  // References attachment ID
  }),
  // Add the attachment table
  attachments: new AttachmentTable()
});

import 'package:powersync/powersync.dart';
import 'package:powersync_core/attachments/attachments.dart';

final schema = Schema([
  Table('users', [
    Column.text('name'),
    Column.text('email'),
    Column.text('photo_id'),  // References attachment ID
  ]),
  AttachmentsQueueTable(),
]);

import PowerSync

let users = Table(
    name: "users",
    columns: [
        Column.text("name"),
        Column.text("email"),
        Column.text("photo_id"),  // References attachment ID
    ]
)

let schema = Schema(
    tables: [
        users,
        // Add the local-only table which stores attachment states
        createAttachmentTable(name: "attachments")
    ]
)

import com.powersync.attachments.createAttachmentsTable
import com.powersync.db.schema.Column
import com.powersync.db.schema.Schema
import com.powersync.db.schema.Table

val users = Table(
    name = "users",
    columns = listOf(
        Column.text("name"),
        Column.text("email"),
        Column.text("photo_id")  // References attachment ID
    )
)

val schema = Schema(
    users,
    // Add the local-only table which stores attachment states
    createAttachmentsTable("attachments")
)

using PowerSync.Common.Attachments;
using PowerSync.Common.DB.Schema;

var users = new Table(
    "users",
    new Dictionary<string, ColumnType>
    {
        ["name"] = ColumnType.Text,
        ["email"] = ColumnType.Text,
        ["photo_id"] = ColumnType.Text,  // References attachment ID
    });

// new Table(typeof(Attachment)) uses the [Table("attachments", LocalOnly = true)] attribute
var schema = new Schema(users, new Table(typeof(Attachment)));

// For web browsers (IndexedDB)
import { IndexDBFileSystemStorageAdapter } from '@powersync/web';
const localStorage = new IndexDBFileSystemStorageAdapter('my-app-files');

// For Node.js/Electron (filesystem)
// import { NodeFileSystemAdapter } from '@powersync/node';
// const localStorage = new NodeFileSystemAdapter('./user-attachments');

// For React Native (Expo or bare React Native)
// Need to install @powersync/attachments-storage-react-native
//
// For Expo projects, also install expo-file-system
// import { ExpoFileSystemStorageAdapter } from '@powersync/attachments-storage-react-native';
// const localStorage = new ExpoFileSystemStorageAdapter();
//
// For bare React Native, also install @dr.pogodin/react-native-fs
// import { ReactNativeFileSystemStorageAdapter } from '@powersync/attachments-storage-react-native';
// const localStorage = new ReactNativeFileSystemStorageAdapter();

// Remote storage adapter (example with signed URLs)
const remoteStorage = {
  async uploadFile(fileData: ArrayBuffer, attachment: AttachmentRecord) {
    // Request signed upload URL from your backend
    const { uploadUrl } = await fetch('/api/attachments/upload-url', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ 
        filename: attachment.filename,
        contentType: attachment.mediaType 
      })
    }).then(r => r.json());
    
    // Upload to cloud storage using signed URL
    await fetch(uploadUrl, {
      method: 'PUT',
      body: fileData,
      headers: { 
        'Content-Type': attachment.mediaType || 'application/octet-stream' 
      }
    });
  },
  
  async downloadFile(attachment: AttachmentRecord): Promise<ArrayBuffer> {
    // Request signed download URL from your backend
    const { downloadUrl } = await fetch(
      `/api/attachments/${attachment.id}/download-url`
    ).then(r => r.json());
    
    // Download from cloud storage
    const response = await fetch(downloadUrl);
    return response.arrayBuffer();
  },
  
  async deleteFile(attachment: AttachmentRecord) {
    // Delete via your backend
    await fetch(`/api/attachments/${attachment.id}`, {
      method: 'DELETE'
    });
  }
};

import 'dart:io';
import 'dart:typed_data';
import 'package:path_provider/path_provider.dart';
import 'package:powersync_core/attachments/attachments.dart';
import 'package:powersync_core/attachments/io.dart';
import 'package:http/http.dart' as http;

// For Flutter (native platforms)
Future<LocalStorage> getLocalStorage() async {
  final appDocDir = await getApplicationDocumentsDirectory();
  final attachmentsDir = Directory('${appDocDir.path}/attachments');
  return IOLocalStorage(attachmentsDir);
}

// Remote storage adapter (example with signed URLs)
class SignedUrlStorageAdapter implements RemoteStorage {
  @override
  Future<void> uploadFile(
    Stream<List<int>> fileData,
    Attachment attachment,
  ) async {
    // Request signed upload URL from your backend
    final response = await http.post(
      Uri.parse('/api/attachments/upload-url'),
      headers: {'Content-Type': 'application/json'},
      body: jsonEncode({
        'filename': attachment.filename,
        'contentType': attachment.mediaType,
      }),
    );
    
    final uploadUrl = jsonDecode(response.body)['uploadUrl'] as String;
    
    // Collect stream data
    final bytes = <int>[];
    await for (final chunk in fileData) {
      bytes.addAll(chunk);
    }
    
    // Upload to cloud storage using signed URL
    await http.put(
      Uri.parse(uploadUrl),
      body: Uint8List.fromList(bytes),
      headers: {
        'Content-Type': attachment.mediaType ?? 'application/octet-stream',
      },
    );
  }
  
  @override
  Future<Stream<List<int>>> downloadFile(Attachment attachment) async {
    // Request signed download URL from your backend
    final response = await http.get(
      Uri.parse('/api/attachments/${attachment.id}/download-url'),
    );
    
    final downloadUrl = jsonDecode(response.body)['downloadUrl'] as String;
    
    // Download from cloud storage
    final httpResponse = await http.get(Uri.parse(downloadUrl));
    return Stream.value(httpResponse.bodyBytes);
  }
  
  @override
  Future<void> deleteFile(Attachment attachment) async {
    // Delete via your backend
    await http.delete(
      Uri.parse('/api/attachments/${attachment.id}'),
    );
  }
}

import Foundation
import PowerSync

// For iOS/macOS (FileManager)
func getAttachmentsDirectoryPath() throws -> String {
    guard let documentsURL = FileManager.default.urls(
        for: .documentDirectory,
        in: .userDomainMask
    ).first else {
        throw PowerSyncAttachmentError.attachmentError("Could not determine attachments directory path")
    }
    return documentsURL.appendingPathComponent("attachments").path
}

let localStorage = FileManagerStorageAdapter()

// Remote storage adapter (example with signed URLs)
class SignedUrlStorageAdapter: RemoteStorageAdapter {
    func uploadFile(fileData: Data, attachment: Attachment) async throws {
        // Request signed upload URL from your backend
        struct UploadUrlResponse: Codable {
            let uploadUrl: String
        }
        
        let requestBody = [
            "filename": attachment.filename,
            "contentType": attachment.mediaType ?? "application/octet-stream"
        ]
        
        var request = URLRequest(url: URL(string: "/api/attachments/upload-url")!)
        request.httpMethod = "POST"
        request.setValue("application/json", forHTTPHeaderField: "Content-Type")
        request.httpBody = try JSONSerialization.data(withJSONObject: requestBody)
        
        let (data, _) = try await URLSession.shared.data(for: request)
        let response = try JSONDecoder().decode(UploadUrlResponse.self, from: data)
        
        // Upload to cloud storage using signed URL
        var uploadRequest = URLRequest(url: URL(string: response.uploadUrl)!)
        uploadRequest.httpMethod = "PUT"
        uploadRequest.setValue(attachment.mediaType ?? "application/octet-stream", forHTTPHeaderField: "Content-Type")
        uploadRequest.httpBody = fileData
        
        let (_, uploadResponse) = try await URLSession.shared.data(for: uploadRequest)
        guard let httpResponse = uploadResponse as? HTTPURLResponse,
              (200...299).contains(httpResponse.statusCode) else {
            throw PowerSyncAttachmentError.generalError("Upload failed")
        }
    }
    
    func downloadFile(attachment: Attachment) async throws -> Data {
        // Request signed download URL from your backend
        struct DownloadUrlResponse: Codable {
            let downloadUrl: String
        }
        
        let request = URLRequest(url: URL(string: "/api/attachments/\(attachment.id)/download-url")!)
        let (data, _) = try await URLSession.shared.data(for: request)
        let response = try JSONDecoder().decode(DownloadUrlResponse.self, from: data)
        
        // Download from cloud storage
        let downloadRequest = URLRequest(url: URL(string: response.downloadUrl)!)
        let (fileData, _) = try await URLSession.shared.data(for: downloadRequest)
        return fileData
    }
    
    func deleteFile(attachment: Attachment) async throws {
        // Delete via your backend
        var request = URLRequest(url: URL(string: "/api/attachments/\(attachment.id)")!)
        request.httpMethod = "DELETE"
        let (_, response) = try await URLSession.shared.data(for: request)
        guard let httpResponse = response as? HTTPURLResponse,
              (200...299).contains(httpResponse.statusCode) else {
            throw PowerSyncAttachmentError.generalError("Delete failed")
        }
    }
}

let remoteStorage = SignedUrlStorageAdapter()

import com.powersync.attachments.LocalStorage
import com.powersync.attachments.RemoteStorage
import com.powersync.attachments.Attachment
import com.powersync.attachments.storage.IOLocalStorageAdapter
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.io.files.Path

// For local storage (uses IOLocalStorageAdapter by default)
// On Android: "${applicationContext.filesDir.canonicalPath}/attachments"
val attachmentsDirectory = Path("attachments").toString()
val localStorage: LocalStorage = IOLocalStorageAdapter()

// Remote storage adapter (example with signed URLs)
val remoteStorage = object : RemoteStorage {
    override suspend fun uploadFile(
        fileData: Flow<ByteArray>,
        attachment: Attachment
    ) {
        // Request signed upload URL from your backend
        val uploadUrl = // ... fetch from your API
        
        // Upload to cloud storage using signed URL
        // Collect the flow and upload
        val bytes = mutableListOf<ByteArray>()
        fileData.collect { bytes.add(it) }
        val allBytes = bytes.flatMap { it.toList() }.toByteArray()
        
        // Upload allBytes to uploadUrl
        // ... your HTTP upload implementation
    }
    
    override suspend fun downloadFile(attachment: Attachment): Flow<ByteArray> {
        // Request signed download URL from your backend
        val downloadUrl = // ... fetch from your API
        
        // Download from cloud storage
        val response = // ... your HTTP download implementation
        return flowOf(response) // or convert your ByteArray to Flow
    }
    
    override suspend fun deleteFile(attachment: Attachment) {
        // Delete via your backend
        // ... your HTTP delete implementation
    }
}

using PowerSync.Common.Attachments;

// Local storage backed by System.IO.File — works on Windows, macOS, Linux, iOS, and Android
var localStorage = new FileManagerLocalStorage(
    Path.Combine(Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData), "myapp", "attachments"));

// Remote storage adapter (example with signed URLs)
public sealed class MyRemoteStorage(HttpClient http) : IRemoteStorageAdapter
{
    public async Task UploadFileAsync(Stream fileData, Attachment attachment)
    {
        // Request a signed upload URL from your backend
        var resp = await http.PostAsJsonAsync("/api/attachments/upload-url",
            new { filename = attachment.Filename });
        var body = await resp.Content.ReadFromJsonAsync<UploadUrlResponse>()
            ?? throw new InvalidOperationException("missing body");

        // Upload directly to cloud storage using the signed URL
        using var content = new StreamContent(fileData);
        content.Headers.ContentType = new(attachment.MediaType ?? "application/octet-stream");
        (await http.PutAsync(body.UploadUrl, content)).EnsureSuccessStatusCode();
    }

    public async Task<Stream> DownloadFileAsync(Attachment attachment)
    {
        // Request a signed download URL from your backend
        var url = await http.GetStringAsync($"/api/attachments/download-url/{attachment.Id}");
        return await http.GetStreamAsync(url);
    }

    public async Task DeleteFileAsync(Attachment attachment)
    {
        // Delete via your backend; 404 is treated as success (file already gone)
        var resp = await http.DeleteAsync($"/api/attachments/{attachment.Id}");
        if (resp.StatusCode != HttpStatusCode.NotFound)
            resp.EnsureSuccessStatusCode();
    }

    private sealed record UploadUrlResponse(string UploadUrl);
}

import { AttachmentQueue } from '@powersync/web';

const attachmentQueue = new AttachmentQueue({
  db: db,  // PowerSync database instance
  localStorage,
  remoteStorage,
  
  // Define which attachments exist in your data model
  watchAttachments: (onUpdate) => {
    db.watch(
      `SELECT photo_id FROM users WHERE photo_id IS NOT NULL`,
      [],
      {
        onResult: async (result) => {
          const attachments = result.rows?._array.map(row => ({
            id: row.photo_id,
            // Names the file in local storage (<id>.jpg). Either fileExtension or filename is required
            fileExtension: 'jpg',
            // MIME type for remote transfer (the Content-Type). Optional, never inferred
            // from fileExtension, and preserved on the downloaded record (JS SDK only)
            mediaType: 'image/jpeg'
          })) ?? [];
          await onUpdate(attachments);
        }
      }
    );
  },
  
  // Optional configuration
  syncIntervalMs: 30000,      // Sync every 30 seconds
  downloadAttachments: true,  // Auto-download referenced files
  archivedCacheLimit: 100     // Keep 100 archived files before cleanup
});

// Start the sync process
await attachmentQueue.startSync();

import 'package:logging/logging.dart';
import 'package:powersync/powersync.dart';
import 'package:powersync_core/attachments/attachments.dart';

final logger = Logger('AttachmentQueue');
late AttachmentQueue attachmentQueue;

Future<void> initializeAttachmentQueue(PowerSyncDatabase db) async {
  attachmentQueue = AttachmentQueue(
    db: db,
    remoteStorage: SignedUrlStorageAdapter(),
    localStorage: await getLocalStorage(),
    
    // Define which attachments exist in your data model
    watchAttachments: () => db.watch('''
      SELECT photo_id as id 
      FROM users 
      WHERE photo_id IS NOT NULL
    ''').map(
      (results) => [
        for (final row in results)
          WatchedAttachmentItem(
            id: row['id'] as String,
            fileExtension: 'jpg',
          )
      ],
    ),
    
    // Optional configuration
    syncInterval: const Duration(seconds: 30),  // Sync every 30 seconds
    downloadAttachments: true,  // Auto-download referenced files
    archivedCacheLimit: 100,  // Keep 100 archived files before cleanup
    logger: logger,
  );
  
  // Start the sync process
  await attachmentQueue.startSync();
}

let attachmentQueue = AttachmentQueue(
    db: db,  // PowerSync database instance
    remoteStorage: remoteStorage,
    attachmentsDirectory: try getAttachmentsDirectoryPath(),
    
    // Define which attachments exist in your data model
    watchAttachments: {
        try db.watch(
            sql: """
                SELECT photo_id 
                FROM users 
                WHERE photo_id IS NOT NULL
            """,
            parameters: [],
            mapper: { cursor in
                try WatchedAttachmentItem(
                    id: cursor.getString(name: "photo_id"),
                    fileExtension: "jpg"
                )
            }
        )
    },
    
    // Optional configuration
    syncInterval: 30.0,      // Sync every 30 seconds
    downloadAttachments: true,  // Auto-download referenced files
    archivedCacheLimit: 100     // Keep 100 archived files before cleanup
)

// Start the sync process
try await attachmentQueue.startSync()

import com.powersync.attachments.AttachmentQueue
import com.powersync.attachments.WatchedAttachmentItem
import com.powersync.db.getString
import kotlinx.coroutines.flow.Flow
import kotlin.time.Duration.Companion.seconds

val attachmentQueue = AttachmentQueue(
    db = db,  // PowerSync database instance
    remoteStorage = remoteStorage,
    attachmentsDirectory = attachmentsDirectory,
    localStorage = localStorage,  // Optional, defaults to IOLocalStorageAdapter()
    
    // Define which attachments exist in your data model
    watchAttachments = {
        db.watch(
            sql = """
                SELECT photo_id 
                FROM users 
                WHERE photo_id IS NOT NULL
            """,
            parameters = null
        ) { cursor ->
            WatchedAttachmentItem(
                id = cursor.getString("photo_id"),
                fileExtension = "jpg"
            )
        }
    },
    
    // Optional configuration
    syncInterval = 30.seconds,      // Sync every 30 seconds
    downloadAttachments = true,      // Auto-download referenced files
    archivedCacheLimit = 100        // Keep 100 archived files before cleanup
)

// Start the sync process
attachmentQueue.startSync()

using PowerSync.Common.Attachments;

var queue = new AttachmentQueue(new AttachmentQueueOptions
{
    Db = powersync,
    LocalStorage = localStorage,
    RemoteStorage = remoteStorage,

    // Define which attachments exist in your data model
    WatchAttachments = ct => WatchProfilePhotos(powersync, ct),

    // Optional configuration
    SyncInterval = TimeSpan.FromSeconds(30),  // Retry failed transfers every 30 seconds
    DownloadAttachments = true,               // Auto-download referenced files
    ArchivedCacheLimit = 100,                 // Keep 100 archived files before cleanup
});

// Start automatic syncing
await queue.StartSyncAsync();

static async IAsyncEnumerable<WatchedAttachmentItem[]> WatchProfilePhotos(
    PowerSyncDatabase db,
    [EnumeratorCancellation] CancellationToken ct)
{
    var stream = db.Watch<UserPhotoRow>(
        "SELECT photo_id FROM users WHERE photo_id IS NOT NULL",
        null,
        new SQLWatchOptions { TriggerImmediately = true, Signal = ct });

    await foreach (var rows in stream.WithCancellation(ct))
        yield return [.. rows.Select(r => new WatchedAttachmentItem(r.photo_id, fileExtension: "jpg"))];
}

internal sealed class UserPhotoRow { public string photo_id { get; set; } = ""; }

// Example: Watching users.photo_id, documents.document_id, and videos.video_id
// Assuming your tables store file extensions
const attachmentQueue = new AttachmentQueue({
  db: db,
  localStorage,
  remoteStorage,
  
  watchAttachments: (onUpdate) => {
    db.watch(
      `SELECT photo_id as id, photo_file_extension as file_extension
       FROM users 
       WHERE photo_id IS NOT NULL
       
       UNION ALL
       
       SELECT document_id as id, document_file_extension as file_extension
       FROM documents 
       WHERE document_id IS NOT NULL
       
       UNION ALL
       
       SELECT video_id as id, video_file_extension as file_extension
       FROM videos 
       WHERE video_id IS NOT NULL`,
      [],
      {
        onResult: async (result) => {
          const attachments = result.rows?._array.map(row => ({
            id: row.id,
            fileExtension: row.file_extension
          })) ?? [];
          await onUpdate(attachments);
        }
      }
    );
  },
  
  // ... other options
});

await attachmentQueue.startSync();

// Example: Watching users.photo_id, documents.document_id, and videos.video_id
// Assuming your tables store file extensions
attachmentQueue = AttachmentQueue(
  db: db,
  remoteStorage: SignedUrlStorageAdapter(),
  localStorage: await getLocalStorage(),
  
  watchAttachments: () => db.watch('''
    SELECT photo_id as id, photo_file_extension as file_extension
    FROM users 
    WHERE photo_id IS NOT NULL
    
    UNION ALL
    
    SELECT document_id as id, document_file_extension as file_extension
    FROM documents 
    WHERE document_id IS NOT NULL
    
    UNION ALL
    
    SELECT video_id as id, video_file_extension as file_extension
    FROM videos 
    WHERE video_id IS NOT NULL
  ''').map(
    (results) => [
      for (final row in results)
        WatchedAttachmentItem(
          id: row['id'] as String,
          fileExtension: row['file_extension'] as String,
        )
    ],
  ),
  
  // ... other options
);

await attachmentQueue.startSync();

// Example: Watching users.photo_id, documents.document_id, and videos.video_id
// Assuming your tables store file extensions
let attachmentQueue = AttachmentQueue(
    db: db,
    remoteStorage: remoteStorage,
    attachmentsDirectory: try getAttachmentsDirectoryPath(),
    
    watchAttachments: {
        try db.watch(
            sql: """
                SELECT photo_id as id, photo_file_extension as file_extension
                FROM users 
                WHERE photo_id IS NOT NULL
                
                UNION ALL
                
                SELECT document_id as id, document_file_extension as file_extension
                FROM documents 
                WHERE document_id IS NOT NULL
                
                UNION ALL
                
                SELECT video_id as id, video_file_extension as file_extension
                FROM videos 
                WHERE video_id IS NOT NULL
            """,
            parameters: [],
            mapper: { cursor in
                try WatchedAttachmentItem(
                    id: cursor.getString(name: "id"),
                    fileExtension: cursor.getString(name: "file_extension")
                )
            }
        )
    },
    
    // ... other options
)

try await attachmentQueue.startSync()

// Example: Watching users.photo_id, documents.document_id, and videos.video_id
// Assuming your tables store file extensions
val attachmentQueue = AttachmentQueue(
    db = db,
    remoteStorage = remoteStorage,
    attachmentsDirectory = attachmentsDirectory,
    localStorage = localStorage,
    
    watchAttachments = {
        db.watch(
            sql = """
                SELECT photo_id as id, photo_file_extension as file_extension
                FROM users 
                WHERE photo_id IS NOT NULL
                
                UNION ALL
                
                SELECT document_id as id, document_file_extension as file_extension
                FROM documents 
                WHERE document_id IS NOT NULL
                
                UNION ALL
                
                SELECT video_id as id, video_file_extension as file_extension
                FROM videos 
                WHERE video_id IS NOT NULL
            """,
            parameters = null
        ) { cursor ->
            WatchedAttachmentItem(
                id = cursor.getString("id"),
                fileExtension = cursor.getString("file_extension")
            )
        }
    },
    
    // ... other options
)

attachmentQueue.startSync()

// Example: watching users.photo_id, documents.document_id, and videos.video_id
// Assuming your tables store file extensions
var queue = new AttachmentQueue(new AttachmentQueueOptions
{
    Db = powersync,
    LocalStorage = localStorage,
    RemoteStorage = remoteStorage,
    WatchAttachments = ct => WatchAllAttachments(powersync, ct),
    // ... other options
});

await queue.StartSyncAsync();

static async IAsyncEnumerable<WatchedAttachmentItem[]> WatchAllAttachments(
    PowerSyncDatabase db,
    [EnumeratorCancellation] CancellationToken ct)
{
    var stream = db.Watch<AttachmentRef>(
        """
        SELECT photo_id AS id, photo_file_extension AS file_extension
        FROM users
        WHERE photo_id IS NOT NULL

        UNION ALL

        SELECT document_id AS id, document_file_extension AS file_extension
        FROM documents
        WHERE document_id IS NOT NULL

        UNION ALL

        SELECT video_id AS id, video_file_extension AS file_extension
        FROM videos
        WHERE video_id IS NOT NULL
        """,
        null,
        new SQLWatchOptions { TriggerImmediately = true, Signal = ct });

    await foreach (var rows in stream.WithCancellation(ct))
        yield return [.. rows.Select(r => new WatchedAttachmentItem(r.id, fileExtension: r.file_extension))];
}

internal sealed class AttachmentRef
{
    public string id { get; set; } = "";
    public string file_extension { get; set; } = "";
}

// Create separate queues for different attachment types
const photoQueue = new AttachmentQueue({
  db: db,
  localStorage,
  remoteStorage,
  watchAttachments: (onUpdate) => {
    db.watch(
      `SELECT photo_id FROM users WHERE photo_id IS NOT NULL`,
      [],
      {
        onResult: async (result) => {
          const attachments = result.rows?._array.map(row => ({
            id: row.photo_id,
            fileExtension: 'jpg',
            mediaType: 'image/jpeg'  // Optional: preserved on the downloaded record
          })) ?? [];
          await onUpdate(attachments);
        }
      }
    );
  },
});

const documentQueue = new AttachmentQueue({
  db: db,
  localStorage,
  remoteStorage,
  watchAttachments: (onUpdate) => {
    db.watch(
      `SELECT document_id FROM documents WHERE document_id IS NOT NULL`,
      [],
      {
        onResult: async (result) => {
          const attachments = result.rows?._array.map(row => ({
            id: row.document_id,
            fileExtension: 'pdf',
            mediaType: 'application/pdf'  // Optional: preserved on the downloaded record
          })) ?? [];
          await onUpdate(attachments);
        }
      }
    );
  },
});

await Promise.all([
  photoQueue.startSync(),
  documentQueue.startSync()
]);

// Create separate queues for different attachment types
final photoQueue = AttachmentQueue(
  db: db,
  remoteStorage: SignedUrlStorageAdapter(),
  localStorage: await getLocalStorage(),
  watchAttachments: () => db.watch('''
    SELECT photo_id as id 
    FROM users 
    WHERE photo_id IS NOT NULL
  ''').map(
    (results) => [
      for (final row in results)
        WatchedAttachmentItem(
          id: row['id'] as String,
          fileExtension: 'jpg',
        )
    ],
  ),
);

final documentQueue = AttachmentQueue(
  db: db,
  remoteStorage: SignedUrlStorageAdapter(),
  localStorage: await getLocalStorage(),
  watchAttachments: () => db.watch('''
    SELECT document_id as id 
    FROM documents 
    WHERE document_id IS NOT NULL
  ''').map(
    (results) => [
      for (final row in results)
        WatchedAttachmentItem(
          id: row['id'] as String,
          fileExtension: 'pdf',
        )
    ],
  ),
);

await Future.wait([
  photoQueue.startSync(),
  documentQueue.startSync(),
]);

// Create separate queues for different attachment types
let photoQueue = AttachmentQueue(
    db: db,
    remoteStorage: remoteStorage,
    attachmentsDirectory: try getAttachmentsDirectoryPath(),
    watchAttachments: {
        try db.watch(
            sql: """
                SELECT photo_id 
                FROM users 
                WHERE photo_id IS NOT NULL
            """,
            parameters: [],
            mapper: { cursor in
                try WatchedAttachmentItem(
                    id: cursor.getString(name: "photo_id"),
                    fileExtension: "jpg"
                )
            }
        )
    }
)

let documentQueue = AttachmentQueue(
    db: db,
    remoteStorage: remoteStorage,
    attachmentsDirectory: try getAttachmentsDirectoryPath(),
    watchAttachments: {
        try db.watch(
            sql: """
                SELECT document_id 
                FROM documents 
                WHERE document_id IS NOT NULL
            """,
            parameters: [],
            mapper: { cursor in
                try WatchedAttachmentItem(
                    id: cursor.getString(name: "document_id"),
                    fileExtension: "pdf"
                )
            }
        )
    }
)

try await photoQueue.startSync()
try await documentQueue.startSync()

// Create separate queues for different attachment types
val photoQueue = AttachmentQueue(
    db = db,
    remoteStorage = remoteStorage,
    attachmentsDirectory = attachmentsDirectory,
    localStorage = localStorage,
    watchAttachments = {
        db.watch(
            sql = """
                SELECT photo_id 
                FROM users 
                WHERE photo_id IS NOT NULL
            """,
            parameters = null
        ) { cursor ->
            WatchedAttachmentItem(
                id = cursor.getString("photo_id"),
                fileExtension = "jpg"
            )
        }
    }
)

val documentQueue = AttachmentQueue(
    db = db,
    remoteStorage = remoteStorage,
    attachmentsDirectory = attachmentsDirectory,
    localStorage = localStorage,
    watchAttachments = {
        db.watch(
            sql = """
                SELECT document_id 
                FROM documents 
                WHERE document_id IS NOT NULL
            """,
            parameters = null
        ) { cursor ->
            WatchedAttachmentItem(
                id = cursor.getString("document_id"),
                fileExtension = "pdf"
            )
        }
    }
)

photoQueue.startSync()
documentQueue.startSync()

// Create separate queues for different attachment types
var photoQueue = new AttachmentQueue(new AttachmentQueueOptions
{
    Db = powersync,
    LocalStorage = localStorage,
    RemoteStorage = remoteStorage,
    WatchAttachments = ct => WatchPhotos(powersync, ct),
});

var documentQueue = new AttachmentQueue(new AttachmentQueueOptions
{
    Db = powersync,
    LocalStorage = localStorage,
    RemoteStorage = remoteStorage,
    WatchAttachments = ct => WatchDocuments(powersync, ct),
});

await Task.WhenAll(
    photoQueue.StartSyncAsync(),
    documentQueue.StartSyncAsync());

static async IAsyncEnumerable<WatchedAttachmentItem[]> WatchPhotos(
    PowerSyncDatabase db,
    [EnumeratorCancellation] CancellationToken ct)
{
    var stream = db.Watch<UserPhotoRow>(
        "SELECT photo_id FROM users WHERE photo_id IS NOT NULL",
        null,
        new SQLWatchOptions { TriggerImmediately = true, Signal = ct });

    await foreach (var rows in stream.WithCancellation(ct))
        yield return [.. rows.Select(r => new WatchedAttachmentItem(r.photo_id, fileExtension: "jpg"))];
}

static async IAsyncEnumerable<WatchedAttachmentItem[]> WatchDocuments(
    PowerSyncDatabase db,
    [EnumeratorCancellation] CancellationToken ct)
{
    var stream = db.Watch<DocumentRow>(
        "SELECT document_id FROM documents WHERE document_id IS NOT NULL",
        null,
        new SQLWatchOptions { TriggerImmediately = true, Signal = ct });

    await foreach (var rows in stream.WithCancellation(ct))
        yield return [.. rows.Select(r => new WatchedAttachmentItem(r.document_id, fileExtension: "pdf"))];
}

internal sealed class DocumentRow { public string document_id { get; set; } = ""; }

async function uploadProfilePhoto(imageBlob: Blob, userId: string) {
  const arrayBuffer = await imageBlob.arrayBuffer();
  
  const attachment = await attachmentQueue.saveFile({
    data: arrayBuffer,
    fileExtension: 'jpg',
    mediaType: 'image/jpeg',
    
    // updateHook runs in same transaction, ensuring atomicity
    updateHook: async (tx, attachment) => {
      await tx.execute(
        'UPDATE users SET photo_id = ? WHERE id = ?',
        [attachment.id, userId]
      );
    }
  });
  
  return attachment;
}

// The queue will:
// 1. Save file locally immediately
// 2. Create attachment record with state QUEUED_UPLOAD
// 3. Update user record in same transaction
// 4. Automatically upload file in background
// 5. Update state to SYNCED when complete

import 'dart:io';
import 'dart:typed_data';
import 'package:powersync_core/attachments/attachments.dart';

Future<Attachment> uploadProfilePhoto(
  File imageFile,
  String userId,
) async {
  final imageBytes = await imageFile.readAsBytes();
  
  final attachment = await attachmentQueue.saveFile(
    data: Stream.value(imageBytes),
    mediaType: 'image/jpeg',
    fileExtension: 'jpg',
    
    // updateHook runs in same transaction, ensuring atomicity
    updateHook: (context, attachment) async {
      await context.execute(
        'UPDATE users SET photo_id = ? WHERE id = ?',
        [attachment.id, userId],
      );
    },
  );
  
  return attachment;
}

// The queue will:
// 1. Save file locally immediately
// 2. Create attachment record with state QUEUED_UPLOAD
// 3. Update user record in same transaction
// 4. Automatically upload file in background
// 5. Update state to SYNCED when complete

func uploadProfilePhoto(imageData: Data, userId: String) async throws -> Attachment {
    let attachment = try await attachmentQueue.saveFile(
        data: imageData,
        mediaType: "image/jpeg",
        fileExtension: "jpg",
        
        // updateHook runs in same transaction, ensuring atomicity
        updateHook: { tx, attachment in
            try tx.execute(
                sql: "UPDATE users SET photo_id = ? WHERE id = ?",
                parameters: [attachment.id, userId]
            )
        }
    )
    
    return attachment
}

// The queue will:
// 1. Save file locally immediately
// 2. Create attachment record with state QUEUED_UPLOAD
// 3. Update user record in same transaction
// 4. Automatically upload file in background
// 5. Update state to SYNCED when complete

import kotlinx.coroutines.flow.flowOf

suspend fun uploadProfilePhoto(imageBytes: ByteArray, userId: String) {
    val attachment = attachmentQueue.saveFile(
        data = flowOf(imageBytes),
        mediaType = "image/jpeg",
        fileExtension = "jpg",
        
        // updateHook runs in same transaction, ensuring atomicity
        updateHook = { tx, attachment ->
            tx.execute(
                "UPDATE users SET photo_id = ? WHERE id = ?",
                listOf(attachment.id, userId)
            )
        }
    )
    
    return attachment
}

// The queue will:
// 1. Save file locally immediately
// 2. Create attachment record with state QUEUED_UPLOAD
// 3. Update user record in same transaction
// 4. Automatically upload file in background
// 5. Update state to SYNCED when complete

async Task UploadProfilePhotoAsync(Stream imageStream, string currentUserId)
{
    var attachment = await queue.SaveFileAsync(
        data: imageStream,
        fileExtension: "jpg",
        mediaType: "image/jpeg",
        // updateHook runs in the same transaction, ensuring atomicity
        updateHook: async (tx, attachment) =>
        {
            await tx.Execute(
                "UPDATE users SET photo_id = ? WHERE id = ?",
                [attachment.Id, currentUserId]);
        });

    // The queue will:
    // 1. Save the file locally immediately
    // 2. Create an attachment record with state QueuedUpload
    // 3. Update the user record in the same transaction
    // 4. Automatically upload the file in the background
    // 5. Update the state to Synced when complete
}

// Downloads happen automatically when watchAttachments references a file

async function getProfilePhotoUri(userId: string): Promise<string | null> {
  const user = await db.get(
    'SELECT photo_id FROM users WHERE id = ?',
    [userId]
  );
  
  if (!user?.photo_id) {
    return null;
  }
  
  const attachment = await db.get(
    'SELECT * FROM attachments WHERE id = ?',
    [user.photo_id]
  );
  
  if (!attachment) {
    return null;
  }
  
  if (attachment.state === 'SYNCED' && attachment.local_uri) {
    return attachment.local_uri;
  }
  
  return null;
}

// Example: Display image in React with watch query
function ProfilePhoto({ userId }: { userId: string }) {
  const [photoUri, setPhotoUri] = useState<string | null>(null);
  
  useEffect(() => {
    const watch = db.watch(
      `SELECT a.local_uri, a.state 
       FROM users u 
       LEFT JOIN attachments a ON a.id = u.photo_id 
       WHERE u.id = ?`,
      [userId],
      {
        onResult: (result) => {
          const row = result.rows?._array[0];
          if (row?.state === 'SYNCED' && row?.local_uri) {
            setPhotoUri(row.local_uri);
          }
        }
      }
    );
    
    return () => watch.close();
  }, [userId]);
  
  if (!photoUri) {
    return <div>Loading photo...</div>;
  }
  
  return <img src={photoUri} alt="Profile" />;
}

import 'package:powersync/powersync.dart';
import 'package:powersync_core/attachments/attachments.dart';

// Downloads happen automatically when watchAttachments references a file
Future<String?> getProfilePhotoUri(
  PowerSyncDatabase db,
  String userId,
) async {
  final user = await db.get(
    'SELECT photo_id FROM users WHERE id = ?',
    [userId],
  );
  
  if (user == null || user['photo_id'] == null) {
    return null;
  }
  
  final attachment = await db.get(
    'SELECT * FROM attachments_queue WHERE id = ?',
    [user['photo_id']],
  );
  
  if (attachment == null) {
    return null;
  }
  
  final state = AttachmentState.fromInt(attachment['state'] as int);
  final localUri = attachment['local_uri'] as String?;
  
  if (state == AttachmentState.synced && localUri != null) {
    // Resolve full path from local storage
    final appDocDir = await getApplicationDocumentsDirectory();
    return '${appDocDir.path}/attachments/$localUri';
  }
  
  return null;
}

// Example: Display image in Flutter with StreamBuilder
StreamBuilder<List<Map<String, dynamic>>>(
  stream: db.watch('''
    SELECT a.local_uri, a.state 
    FROM users u 
    LEFT JOIN attachments_queue a ON a.id = u.photo_id 
    WHERE u.id = ?
  ''').map((results) => results.toList()),
  builder: (context, snapshot) {
    if (!snapshot.hasData || snapshot.data!.isEmpty) {
      return const CircularProgressIndicator();
    }
    
    final row = snapshot.data!.first;
    final state = AttachmentState.fromInt(row['state'] as int);
    final localUri = row['local_uri'] as String?;
    
    if (state == AttachmentState.synced && localUri != null) {
      // Load and display image
      return Image.file(File(localUri));
    }
    
    return const Text('Loading photo...');
  },
)

// Downloads happen automatically when watchAttachments references a file
func getProfilePhotoUri(userId: String) async throws -> String? {
    guard let user = try await db.getOptional(
        sql: "SELECT photo_id FROM users WHERE id = ?",
        parameters: [userId],
        mapper: { cursor in
            try cursor.getStringOptional(name: "photo_id")
        }
    ), let photoId = user else {
        return nil
    }
    
    guard let attachment = try await db.getOptional(
        sql: "SELECT * FROM attachments WHERE id = ?",
        parameters: [photoId],
        mapper: { cursor in
            try Attachment.fromCursor(cursor)
        }
    ) else {
        return nil
    }
    
    if attachment.state == .synced, let localUri = attachment.localUri {
        return localUri
    }
    
    return nil
}

// Example: Display image in SwiftUI with watch query
struct ProfilePhotoView: View {
    let userId: String
    @State private var photoUri: String?
    
    var body: some View {
        Group {
            if let photoUri = photoUri {
                AsyncImage(url: URL(fileURLWithPath: photoUri)) { image in
                    image.resizable()
                } placeholder: {
                    ProgressView()
                }
            } else {
                Text("Loading photo...")
            }
        }
        .task {
            do {
                for try await results in try db.watch(
                    sql: """
                        SELECT a.local_uri, a.state 
                        FROM users u 
                        LEFT JOIN attachments a ON a.id = u.photo_id 
                        WHERE u.id = ?
                    """,
                    parameters: [userId],
                    mapper: { cursor in
                        (
                            state: try AttachmentState.from(cursor.getInt(name: "state")),
                            localUri: try cursor.getStringOptional(name: "local_uri")
                        )
                    }
                ) {
                    if let first = results.first,
                       first.state == .synced,
                       let localUri = first.localUri {
                        photoUri = localUri
                    }
                }
            } catch {
                print("Error watching photo: \(error)")
            }
        }
    }
}

import com.powersync.attachments.AttachmentState
import com.powersync.db.getString
import com.powersync.db.getStringOptional
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map

// Downloads happen automatically when watchAttachments references a file
suspend fun getProfilePhotoUri(userId: String): String? {
    val user = db.get(
        "SELECT photo_id FROM users WHERE id = ?",
        listOf(userId)
    ) { cursor ->
        cursor.getStringOptional("photo_id")
    }
    
    if (user == null) {
        return null
    }
    
    val attachment = db.get(
        "SELECT * FROM attachments WHERE id = ?",
        listOf(user)
    ) { cursor ->
        com.powersync.attachments.Attachment.fromCursor(cursor)
    }
    
    if (attachment == null) {
        return null
    }
    
    if (attachment.state == AttachmentState.SYNCED && attachment.localUri != null) {
        return attachment.localUri
    }
    
    return null
}

// Example: Watch attachment state in Compose/UI
fun watchProfilePhoto(userId: String): Flow<String?> {
    return db.watch(
        sql = """
            SELECT a.local_uri, a.state 
            FROM users u 
            LEFT JOIN attachments a ON a.id = u.photo_id 
            WHERE u.id = ?
        """,
        parameters = listOf(userId)
    ) { cursor ->
        val state = AttachmentState.fromLong(cursor.getLong("state"))
        val localUri = cursor.getStringOptional("local_uri")
        
        if (state == AttachmentState.SYNCED && localUri != null) {
            localUri
        } else {
            null
        }
    }.map { results -> results.firstOrNull() }
}

// Downloads happen automatically when WatchAttachments references a file

// Reactively yield the local file path whenever the attachment reaches the Synced state
static async IAsyncEnumerable<string?> WatchProfilePhotoUri(
    PowerSyncDatabase db,
    string userId,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    var stream = db.Watch<PhotoState>(
        """
        SELECT a.local_uri, a.state
        FROM users u
        LEFT JOIN attachments a ON a.id = u.photo_id
        WHERE u.id = ?
        """,
        [userId],
        new SQLWatchOptions { TriggerImmediately = true, Signal = ct });

    await foreach (var rows in stream.WithCancellation(ct))
    {
        var row = rows.FirstOrDefault();
        yield return row?.state == (long)AttachmentState.Synced ? row.local_uri : null;
    }
}

internal sealed class PhotoState
{
    public string? local_uri { get; set; }
    public long state { get; set; }
}

async function deleteProfilePhoto(userId: string, photoId: string) {
  await attachmentQueue.deleteFile({
    id: photoId,
    
    // updateHook ensures atomic deletion
    updateHook: async (tx, attachment) => {
      await tx.execute(
        'UPDATE users SET photo_id = NULL WHERE id = ?',
        [userId]
      );
    }
  });
  
  console.log('Photo queued for deletion');
  // The queue will:
  // 1. Delete from remote storage
  // 2. Delete local file
  // 3. Remove attachment record
}

// Alternative: Remove reference and let queue archive it automatically
async function removePhotoReference(userId: string) {
  await db.execute(
    'UPDATE users SET photo_id = NULL WHERE id = ?',
    [userId]
  );
  
  // The watchAttachments callback will detect this change
  // The queue will automatically archive the unreferenced attachment
  // After reaching archivedCacheLimit, it will be deleted
}

Future<void> deleteProfilePhoto(
  String userId,
  String photoId,
) async {
  await attachmentQueue.deleteFile(
    attachmentId: photoId,
    
    // updateHook ensures atomic deletion
    updateHook: (context, attachment) async {
      await context.execute(
        'UPDATE users SET photo_id = NULL WHERE id = ?',
        [userId],
      );
    },
  );
  
  print('Photo queued for deletion');
  // The queue will:
  // 1. Delete from remote storage
  // 2. Delete local file
  // 3. Remove attachment record
}

// Alternative: Remove reference and let queue archive it automatically
Future<void> removePhotoReference(
  PowerSyncDatabase db,
  String userId,
) async {
  await db.execute(
    'UPDATE users SET photo_id = NULL WHERE id = ?',
    [userId],
  );
  
  // The watchAttachments callback will detect this change
  // The queue will automatically archive the unreferenced attachment
  // After reaching archivedCacheLimit, it will be deleted
}

func deleteProfilePhoto(userId: String, photoId: String) async throws {
    try await attachmentQueue.deleteFile(
        attachmentId: photoId,
        
        // updateHook ensures atomic deletion
        updateHook: { tx, attachment in
            try tx.execute(
                sql: "UPDATE users SET photo_id = NULL WHERE id = ?",
                parameters: [userId]
            )
        }
    )
    
    print("Photo queued for deletion")
    // The queue will:
    // 1. Delete from remote storage
    // 2. Delete local file
    // 3. Remove attachment record
}

// Alternative: Remove reference and let queue archive it automatically
func removePhotoReference(userId: String) async throws {
    try await db.execute(
        sql: "UPDATE users SET photo_id = NULL WHERE id = ?",
        parameters: [userId]
    )
    
    // The watchAttachments callback will detect this change
    // The queue will automatically archive the unreferenced attachment
    // After reaching archivedCacheLimit, it will be deleted
}

suspend fun deleteProfilePhoto(userId: String, photoId: String) {
    attachmentQueue.deleteFile(
        attachmentId = photoId,
        
        // updateHook ensures atomic deletion
        updateHook = { tx, attachment ->
            tx.execute(
                "UPDATE users SET photo_id = NULL WHERE id = ?",
                listOf(userId)
            )
        }
    )
    
    // The queue will:
    // 1. Delete from remote storage
    // 2. Delete local file
    // 3. Remove attachment record
}

// Alternative: Remove reference and let queue archive it automatically
suspend fun removePhotoReference(userId: String) {
    db.writeTransaction { tx ->
        tx.execute(
            "UPDATE users SET photo_id = NULL WHERE id = ?",
            listOf(userId)
        )
    }
    
    // The watchAttachments callback will detect this change
    // The queue will automatically archive the unreferenced attachment
    // After reaching archivedCacheLimit, it will be deleted
}

async Task DeleteProfilePhotoAsync(string userId, string photoId)
{
    await queue.DeleteFileAsync(
        id: photoId,
        // updateHook ensures the data model update and deletion are atomic
        updateHook: async (tx, attachment) =>
        {
            await tx.Execute(
                "UPDATE users SET photo_id = NULL WHERE id = ?",
                [userId]);
        });

    // The queue will:
    // 1. Delete from remote storage
    // 2. Delete the local file
    // 3. Remove the attachment record
}

// Alternative: remove the reference and let the queue archive it automatically
async Task RemovePhotoReferenceAsync(string userId)
{
    await powersync.Execute(
        "UPDATE users SET photo_id = NULL WHERE id = ?",
        [userId]);

    // The WatchAttachments callback detects this change.
    // The queue archives the unreferenced attachment automatically.
    // Once the cache limit is reached, archived attachments are deleted.
}

import { AttachmentErrorHandler } from '@powersync/web';

const errorHandler: AttachmentErrorHandler = {
  async onDownloadError(attachment, error) {
    console.error(`Download failed: ${attachment.filename}`, error);
    
    // Return true to retry, false to archive
    if (error.message.includes('404')) {
      return false; // File doesn't exist, don't retry
    }
    return true; // Retry on network errors
  },
  
  async onUploadError(attachment, error) {
    console.error(`Upload failed: ${attachment.filename}`, error);
    return true; // Always retry uploads
  },
  
  async onDeleteError(attachment, error) {
    console.error(`Delete failed: ${attachment.filename}`, error);
    return true; // Retry deletes
  }
};

const queue = new AttachmentQueue({
  // ... other options
  errorHandler
});

import 'package:powersync_core/attachments/attachments.dart';

final errorHandler = AttachmentErrorHandler(
  onDownloadError: (attachment, exception, stackTrace) async {
    print('Download failed: ${attachment.filename}');
    print('Error: $exception');
    
    // Return true to retry, false to archive
    if (exception.toString().contains('404')) {
      return false; // File doesn't exist, don't retry
    }
    return true; // Retry on network errors
  },
  
  onUploadError: (attachment, exception, stackTrace) async {
    print('Upload failed: ${attachment.filename}');
    print('Error: $exception');
    return true; // Always retry uploads
  },
  
  onDeleteError: (attachment, exception, stackTrace) async {
    print('Delete failed: ${attachment.filename}');
    print('Error: $exception');
    return true; // Retry deletes
  },
);

final queue = AttachmentQueue(
  // ... other options
  errorHandler: errorHandler,
);