feat: add appMetadata for service logs

Author: stevensJourney

CHANGELOG.md

@@ -4,6 +4,17 @@
 
 - Sync options: `newClientImplementation` is now the default.
 - Make `androidx.sqlite:sqlite-bundled` an API dependency of `:core` to avoid toolchain warnings.
+- Add `appMetadata` parameter to `PowerSyncDatabase.connect()` to include application metadata in sync requests. This metadata is merged into sync requests and displayed in PowerSync service logs.
+
+```kotlin
+database.connect(
+    connector = connector,
+    appMetadata = mapOf(
+        "appVersion" to "1.0.0",
+        "deviceId" to "device456"
+    )
+)
+```
 
 ## 1.8.1
 

common/src/commonMain/kotlin/com/powersync/PowerSyncDatabase.kt

@@ -80,6 +80,7 @@ public interface PowerSyncDatabase : Queries {
      *  Use @param [crudThrottleMs] to specify the time between CRUD operations. Defaults to 1000ms.
      *  Use @param [retryDelayMs] to specify the delay between retries after failure. Defaults to 5000ms.
      *  Use @param [params] to specify sync parameters from the client.
+     *  Use @param [appMetadata] to specify application metadata that will be displayed in PowerSync service logs.
      *
      *  Example usage:
      *  ```
@@ -91,11 +92,17 @@ public interface PowerSyncDatabase : Queries {
      *      )
      *   )
      *
+     *  val appMetadata = mapOf(
+     *      "appVersion" to "1.0.0",
+     *      "deviceId" to "device456"
+     *  )
+     *
      *  connect(
      *      connector = connector,
      *      crudThrottleMs = 2000L,
      *      retryDelayMs = 10000L,
-     *      params = params
+     *      params = params,
+     *      appMetadata = appMetadata
      *  )
      *  ```
      */
@@ -106,6 +113,7 @@ public interface PowerSyncDatabase : Queries {
         retryDelayMs: Long = 5000L,
         params: Map<String, JsonParam?> = emptyMap(),
         options: SyncOptions = SyncOptions(),
+        appMetadata: Map<String, String> = emptyMap(),
     )
 
     /**
@@ -272,7 +280,8 @@ public interface PowerSyncDatabase : Queries {
             val logger = generateLogger(logger)
             // Since this returns a fresh in-memory database every time, use a fresh group to avoid warnings about the
             // same database being opened multiple times.
-            val collection = ActiveDatabaseGroup.GroupsCollection().referenceDatabase(logger, "test")
+            val collection =
+                ActiveDatabaseGroup.GroupsCollection().referenceDatabase(logger, "test")
 
             return openedWithGroup(
                 SingleConnectionPool(factory.openInMemoryConnection()),

common/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt

@@ -71,6 +71,8 @@ internal sealed interface PowerSyncControlArguments {
         val includeDefaults: Boolean,
         @SerialName("active_streams")
         val activeStreams: List<StreamKey>,
+        @SerialName("app_metadata")
+        val appMetadata: Map<String, String>,
     ) : PowerSyncControlArguments {
         override val sqlArguments: Pair<String, Any?>
             get() = "start" to JsonUtil.json.encodeToString(this)
@@ -109,7 +111,8 @@ internal sealed interface PowerSyncControlArguments {
     class UpdateSubscriptions(
         activeStreams: List<StreamKey>,
     ) : PowerSyncControlArguments {
-        override val sqlArguments: Pair<String, Any?> = "update_subscriptions" to JsonUtil.json.encodeToString(activeStreams)
+        override val sqlArguments: Pair<String, Any?> =
+            "update_subscriptions" to JsonUtil.json.encodeToString(activeStreams)
     }
 }
 

common/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -142,6 +142,7 @@ internal class PowerSyncDatabaseImpl(
         retryDelayMs: Long,
         params: Map<String, JsonParam?>,
         options: SyncOptions,
+        appMetadata: Map<String, String>,
     ) {
         waitReady()
         mutex.withLock {

common/src/commonMain/kotlin/com/powersync/sync/StreamingSync.kt

@@ -63,6 +63,7 @@ import kotlinx.io.readByteArray
 import kotlinx.io.readIntLe
 import kotlinx.serialization.json.JsonElement
 import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
 import kotlinx.serialization.json.encodeToJsonElement
 import kotlin.experimental.ExperimentalObjCRefinement
 import kotlin.native.HiddenFromObjC
@@ -115,6 +116,7 @@ internal class StreamingSyncClient(
     private val options: SyncOptions,
     private val schema: Schema,
     private val activeSubscriptions: StateFlow<List<SubscriptionGroup>>,
+    private val appMetadata: Map<String, String>,
 ) {
     private var isUploadingCrud = AtomicReference<PendingCrudUpload?>(null)
     private var completedCrudUploads = Channel<Unit>(onBufferOverflow = BufferOverflow.DROP_OLDEST)
@@ -176,7 +178,13 @@ internal class StreamingSyncClient(
                 status.update { copy(downloadError = e) }
             } finally {
                 if (!result.hideDisconnectStateAndReconnectImmediately) {
-                    status.update { copy(connected = false, connecting = true, downloading = false) }
+                    status.update {
+                        copy(
+                            connected = false,
+                            connecting = true,
+                            downloading = false,
+                        )
+                    }
                     delay(retryDelayMs)
                 }
             }
@@ -397,6 +405,7 @@ internal class StreamingSyncClient(
                     schema = schema.toSerializable(),
                     includeDefaults = options.includeDefaultStreams,
                     activeStreams = subscriptions.map { it.key },
+                    appMetadata = appMetadata,
                 ),
             )
 
@@ -405,7 +414,9 @@ internal class StreamingSyncClient(
                     activeSubscriptions.collect {
                         if (subscriptions !== it) {
                             subscriptions = it
-                            controlInvocations.send(PowerSyncControlArguments.UpdateSubscriptions(activeSubscriptions.value.map { it.key }))
+                            controlInvocations.send(
+                                PowerSyncControlArguments.UpdateSubscriptions(activeSubscriptions.value.map { it.key }),
+                            )
                         }
                     }
                 }
@@ -525,10 +536,52 @@ internal class StreamingSyncClient(
         }
 
         private suspend fun connect(start: Instruction.EstablishSyncStream) {
-            receiveTextOrBinaryLines(start.request).collect {
+            // Merge local appMetadata from StreamingSyncClient into the request before sending
+            val mergedRequest = mergeAppMetadata(start.request)
+            receiveTextOrBinaryLines(mergedRequest).collect {
                 controlInvocations.send(it)
             }
         }
+
+        /**
+         * FIXME, the Rust implementation does not yet pass app_metadata to the sync instruction
+         *
+         * Merges local appMetadata into the request JsonObject.
+         * If the request already has app_metadata, the local appMetadata will be merged into it
+         * (with local values taking precedence for duplicate keys).
+         */
+        private fun mergeAppMetadata(request: JsonObject): JsonObject {
+            if (appMetadata.isEmpty()) {
+                return request
+            }
+
+            // Convert local appMetadata to JsonObject
+            val localAppMetadataJson =
+                JsonObject(
+                    appMetadata.mapValues { (_, value) -> JsonPrimitive(value) },
+                )
+
+            // Get existing app_metadata from request, if any
+            val existingAppMetadata =
+                request["app_metadata"] as? JsonObject ?: JsonObject(emptyMap())
+
+            // Merge: existing first, then local (local takes precedence)
+            val mergedAppMetadata =
+                JsonObject(
+                    buildMap {
+                        putAll(existingAppMetadata)
+                        putAll(localAppMetadataJson)
+                    },
+                )
+
+            // Create new request with merged app_metadata
+            return JsonObject(
+                buildMap {
+                    putAll(request)
+                    put("app_metadata", mergedAppMetadata)
+                },
+            )
+        }
     }
 
     @LegacySyncImplementation
@@ -566,6 +619,7 @@ internal class StreamingSyncClient(
                         },
                     clientId = clientId!!,
                     parameters = params,
+                    appMetadata = appMetadata,
                 )
 
             lateinit var receiveLines: Job

common/src/commonMain/kotlin/com/powersync/sync/StreamingSyncRequest.kt

@@ -12,6 +12,7 @@ internal data class StreamingSyncRequest(
     @SerialName("include_checksum") val includeChecksum: Boolean = true,
     @SerialName("client_id") val clientId: String,
     val parameters: JsonObject = JsonObject(mapOf()),
+    @SerialName("app_metadata") val appMetadata: Map<String, String>,
 ) {
     @SerialName("raw_data")
     private val rawData: Boolean = true