wip: Android Attachments Demo

connectors/supabase/build.gradle.kts

@@ -36,6 +36,7 @@ kotlin {
             implementation(libs.kotlinx.coroutines.core)
             implementation(libs.supabase.client)
             api(libs.supabase.auth)
+            api(libs.supabase.storage)
         }
     }
 }

connectors/supabase/src/commonMain/kotlin/com/powersync/connector/supabase/SupabaseConnector.kt

@@ -17,6 +17,9 @@ import io.github.jan.supabase.auth.user.UserSession
 import io.github.jan.supabase.createSupabaseClient
 import io.github.jan.supabase.postgrest.Postgrest
 import io.github.jan.supabase.postgrest.from
+import io.github.jan.supabase.storage.BucketApi
+import io.github.jan.supabase.storage.Storage
+import io.github.jan.supabase.storage.storage
 import io.ktor.client.plugins.HttpSend
 import io.ktor.client.plugins.plugin
 import io.ktor.client.statement.bodyAsText
@@ -31,6 +34,7 @@ import kotlinx.serialization.json.Json
 public class SupabaseConnector(
     public val supabaseClient: SupabaseClient,
     public val powerSyncEndpoint: String,
+    private val storageBucket: String? = null,
 ) : PowerSyncBackendConnector() {
     private var errorCode: String? = null
 
@@ -52,17 +56,29 @@ public class SupabaseConnector(
             }
     }
 
+    public fun storageBucket(): BucketApi {
+        if (storageBucket == null) {
+            throw Exception("No bucket has been specified")
+        }
+        return supabaseClient.storage[storageBucket]
+    }
+
     public constructor(
         supabaseUrl: String,
         supabaseKey: String,
         powerSyncEndpoint: String,
+        storageBucket: String? = null,
     ) : this(
         supabaseClient =
             createSupabaseClient(supabaseUrl, supabaseKey) {
                 install(Auth)
                 install(Postgrest)
+                if (storageBucket != null) {
+                    install(Storage)
+                }
             },
         powerSyncEndpoint = powerSyncEndpoint,
+        storageBucket = storageBucket,
     )
 
     init {
@@ -81,7 +97,10 @@ public class SupabaseConnector(
                 val responseText = response.bodyAsText()
 
                 try {
-                    val error = Json { coerceInputValues = true }.decodeFromString<Map<String, String?>>(responseText)
+                    val error =
+                        Json { coerceInputValues = true }.decodeFromString<Map<String, String?>>(
+                            responseText,
+                        )
                     errorCode = error["code"]
                 } catch (e: Exception) {
                     Logger.e("Failed to parse error response: $e")
@@ -139,7 +158,9 @@ public class SupabaseConnector(
             check(supabaseClient.auth.sessionStatus.value is SessionStatus.Authenticated) { "Supabase client is not authenticated" }
 
             // Use Supabase token for PowerSync
-            val session = supabaseClient.auth.currentSessionOrNull() ?: error("Could not fetch Supabase credentials")
+            val session =
+                supabaseClient.auth.currentSessionOrNull()
+                    ?: error("Could not fetch Supabase credentials")
 
             check(session.user != null) { "No user data" }
 

core/src/androidMain/kotlin/com/powersync/attachments/storage/AbstractLocalStorageAdapter.android.kt

@@ -1,9 +1,12 @@
 package com.powersync.attachments.storage
 
 import android.os.Environment
+import android.os.Environment.DIRECTORY_DOCUMENTS
 import com.powersync.attachments.LocalStorageAdapter
 
+import okio.FileSystem
+
 @Suppress("EXPECT_ACTUAL_CLASSIFIERS_ARE_IN_BETA_WARNING")
 public actual abstract class AbstractLocalStorageAdapter : LocalStorageAdapter {
-    actual override fun getUserStorageDirectory(): String = Environment.getDataDirectory().absolutePath
+    actual override fun getUserStorageDirectory(): String = FileSystem.SYSTEM_TEMPORARY_DIRECTORY.normalized().toString()
 }

core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt

@@ -80,7 +80,7 @@ class AttachmentsTest {
                         archivedCacheLimit = 0,
                     )
 
-                queue.start()
+                queue.startSync()
 
                 // Monitor the attachments table for testing
                 val attachmentQuery =
@@ -184,7 +184,7 @@ class AttachmentsTest {
                         archivedCacheLimit = 0,
                     )
 
-                queue.start()
+                queue.startSync()
 
                 // Monitor the attachments table for testing
                 val attachmentQuery =
@@ -285,7 +285,7 @@ class AttachmentsTest {
                         archivedCacheLimit = 10,
                     )
 
-                queue.start()
+                queue.startSync()
 
                 // Monitor the attachments table for testing
                 val attachmentQuery =
@@ -411,7 +411,7 @@ class AttachmentsTest {
                             },
                     )
 
-                queue.start()
+                queue.startSync()
 
                 // Monitor the attachments table for testing
                 val attachmentQuery =

core/src/commonMain/kotlin/com/powersync/attachments/AbstractAttachmentQueue.kt

@@ -88,6 +88,10 @@ public abstract class AbstractAttachmentQueue(
      * Records are deleted once the number of items exceeds this value.
      */
     private val archivedCacheLimit: Long = 100,
+    /**
+     * Throttles remote sync operations triggering
+     */
+    private val syncThrottleDuration: Duration = 1.seconds,
     /**
      * Creates a list of subdirectories in the {attachmentDirectoryName} directory
      */
@@ -133,6 +137,7 @@ public abstract class AbstractAttachmentQueue(
             ::getLocalUri,
             errorHandler,
             logger,
+            syncThrottleDuration,
         )
 
     private var syncStatusJob: Job? = null
@@ -147,7 +152,7 @@ public abstract class AbstractAttachmentQueue(
      * 3. Adding trigger to run uploads, downloads, and deletes when device is online after being offline
      */
     @Throws(PowerSyncException::class, CancellationException::class)
-    public suspend fun start(): Unit =
+    public suspend fun startSync(): Unit =
         runWrappedSuspending {
             mutex.withLock {
                 if (closed) {
@@ -353,7 +358,7 @@ public abstract class AbstractAttachmentQueue(
         data: Flow<ByteArray>,
         mediaType: String,
         fileExtension: String?,
-        updateHook: (context: ConnectionContext, attachment: Attachment) -> Unit,
+        updateHook: ((context: ConnectionContext, attachment: Attachment) -> Unit)? = null,
     ): Attachment =
         runWrappedSuspending {
             val id = db.get("SELECT uuid()") { it.getString(0)!! }
@@ -369,22 +374,24 @@ public abstract class AbstractAttachmentQueue(
              * assignment should happen in the same transaction.
              */
             db.writeTransaction { tx ->
-                val attachment =
-                    attachmentsService.upsertAttachment(
-                        Attachment(
-                            id = id,
-                            filename = filename,
-                            size = fileSize,
-                            mediaType = mediaType,
-                            state = AttachmentState.QUEUED_UPLOAD.ordinal,
-                            localUri = localUri,
-                        ),
-                        tx,
-                    )
-
-                // Allow consumers to set relationships to this attachment id
-                updateHook(tx, attachment)
-                return@writeTransaction attachment
+                val attachment =  Attachment(
+                    id = id,
+                    filename = filename,
+                    size = fileSize,
+                    mediaType = mediaType,
+                    state = AttachmentState.QUEUED_UPLOAD.ordinal,
+                    localUri = localUri,
+                )
+
+                /**
+                 * Allow consumers to set relationships to this attachment id
+                  */
+                updateHook?.invoke(tx, attachment)
+
+                return@writeTransaction attachmentsService.upsertAttachment(
+                    attachment,
+                    tx,
+                )
             }
         }
 
@@ -396,15 +403,21 @@ public abstract class AbstractAttachmentQueue(
      * This method can be overriden for custom behaviour.
      */
     @Throws(PowerSyncException::class, CancellationException::class)
-    public open suspend fun deleteFile(attachmentId: String): Attachment =
+    public open suspend fun deleteFile(attachmentId: String,
+                                       updateHook: ((context: ConnectionContext, attachment: Attachment) -> Unit)? = null,
+                                       ): Attachment =
         runWrappedSuspending {
             val attachment =
                 attachmentsService.getAttachment(attachmentId)
                     ?: throw Exception("Attachment record with id $attachmentId was not found.")
 
-            return@runWrappedSuspending attachmentsService.saveAttachment(
-                attachment.copy(state = AttachmentState.QUEUED_DELETE.ordinal),
-            )
+            db.writeTransaction { tx ->
+                updateHook?.invoke(tx, attachment)
+                return@writeTransaction attachmentsService.upsertAttachment(
+                    attachment.copy(state = AttachmentState.QUEUED_DELETE.ordinal),
+                    tx,
+                )
+            }
         }
 
     /**

core/src/commonMain/kotlin/com/powersync/attachments/README.md

@@ -141,7 +141,7 @@ val remote = object : RemoteStorageAdapter() {
 
 ```
 
-5. Instantiate a new `AttachmentQueue` and call `start()` to start syncing attachments.
+5. Instantiate a new `AttachmentQueue` and call `startSync()` to start syncing attachments.
 
 ```kotlin
     val queue =
@@ -150,7 +150,7 @@ val remote = object : RemoteStorageAdapter() {
         remoteStorage = remote,
     )
 
-queue.start()
+queue.startSync()
 ```
 
 7. Finally, to create an attachment and add it to the queue, call `saveFile()`. This method will

core/src/commonMain/kotlin/com/powersync/attachments/storage/IOLocalStorageAdapter.kt

@@ -2,11 +2,13 @@ package com.powersync.attachments.storage
 
 import com.powersync.attachments.storage.AbstractLocalStorageAdapter
 import com.powersync.db.runWrappedSuspending
+import io.ktor.utils.io.core.readBytes
 import io.ktor.utils.io.core.remaining
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.IO
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.withContext
 import kotlinx.io.Buffer
 import kotlinx.io.buffered
@@ -48,23 +50,19 @@ public class IOLocalStorageAdapter : AbstractLocalStorageAdapter() {
         mediaType: String?,
     ): Flow<ByteArray> =
         flow {
-            withContext(Dispatchers.IO) {
                 SystemFileSystem.source(Path(filePath)).use { source ->
                     source.buffered().use { bufferedSource ->
-                        val bufferSize = 8192L // Read in 8KB chunks
-                        while (bufferedSource.remaining > 0) {
-                            val byteCount =
-                                min(
-                                    bufferedSource.remaining,
-                                    bufferSize,
-                                )
-                            val bytesRead = bufferedSource.readByteArray(byteCount.toInt())
-                            emit(bytesRead)
-                        }
+                        var remaining  = 0L
+                        val bufferSize = 8192L
+                        do {
+                            bufferedSource.request(bufferSize)
+                            remaining  = bufferedSource.remaining
+                            emit(bufferedSource.readBytes(remaining.toInt()))
+                        } while (remaining > 0)
+
                     }
                 }
-            }
-        }
+        }.flowOn(Dispatchers.IO)
 
     public override suspend fun deleteFile(filePath: String): Unit =
         runWrappedSuspending {

core/src/commonMain/kotlin/com/powersync/attachments/sync/SyncingService.kt

@@ -8,6 +8,7 @@ import com.powersync.attachments.AttachmentState
 import com.powersync.attachments.LocalStorageAdapter
 import com.powersync.attachments.RemoteStorageAdapter
 import com.powersync.attachments.SyncErrorHandler
+import com.powersync.utils.throttle
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -23,6 +24,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
 
 /**
  * Service used to sync attachments between local and remote storage
@@ -34,6 +36,7 @@ internal class SyncingService(
     private val getLocalUri: suspend (String) -> String,
     private val errorHandler: SyncErrorHandler?,
     private val logger: Logger,
+    private val syncThrottle: Duration = 5.seconds,
 ) {
     private val scope = CoroutineScope(Dispatchers.IO)
     private val mutex = Mutex()
@@ -60,6 +63,7 @@ internal class SyncingService(
                     // while we are processing. We will always process on the trailing edge.
                     // This buffer operation should automatically be applied to all merged sources.
                     .buffer(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+                    .throttle(syncThrottle.inWholeMilliseconds)
                     .collect {
                         /**
                          * Gets and performs the operations for active attachments which are

demos/android-supabase-todolist/app/build.gradle.kts

@@ -13,9 +13,10 @@ if (localPropertiesFile.exists()) {
     localPropertiesFile.inputStream().use { localProperties.load(it) }
 }
 
-fun getLocalProperty(key: String, defaultValue: String): String {
-    return localProperties.getProperty(key, defaultValue)
-}
+fun getLocalProperty(
+    key: String,
+    defaultValue: String,
+): String = localProperties.getProperty(key, defaultValue)
 
 android {
     namespace = "com.powersync.androidexample"
@@ -38,7 +39,16 @@ android {
         }
 
         buildConfigField("String", "SUPABASE_URL", "\"${getLocalProperty("SUPABASE_URL", "")}\"")
-        buildConfigField("String", "SUPABASE_ANON_KEY", "\"${getLocalProperty("SUPABASE_ANON_KEY", "")}\"")
+        buildConfigField(
+            "String",
+            "SUPABASE_ANON_KEY",
+            "\"${getLocalProperty("SUPABASE_ANON_KEY", "")}\"",
+        )
+        buildConfigField(
+            "String",
+            "SUPABASE_ATTACHMENT_BUCKET",
+            "\"${getLocalProperty("SUPABASE_ATTACHMENT_BUCKET", "")}\"",
+        )
         buildConfigField("String", "POWERSYNC_URL", "\"${getLocalProperty("POWERSYNC_URL", "")}\"")
     }
 
@@ -47,7 +57,7 @@ android {
             isMinifyEnabled = false
             proguardFiles(
                 getDefaultProguardFile("proguard-android-optimize.txt"),
-                "proguard-rules.pro"
+                "proguard-rules.pro",
             )
         }
     }
@@ -96,4 +106,5 @@ dependencies {
     implementation(libs.uuid)
     implementation(libs.kermit)
     implementation(libs.androidx.material.icons.extended)
-}
+    implementation("androidx.activity:activity-ktx:1.9.0")
+}