Add ResourceConsolidator

engine/src/androidTest/java/com/google/android/fhir/db/impl/DatabaseImplTest.kt

@@ -35,6 +35,8 @@ import com.google.android.fhir.search.Search
 import com.google.android.fhir.search.StringFilterModifier
 import com.google.android.fhir.search.getQuery
 import com.google.android.fhir.search.has
+import com.google.android.fhir.sync.UrlUploadRequest
+import com.google.android.fhir.sync.upload.UploadMode
 import com.google.android.fhir.testing.assertJsonArrayEqualsIgnoringOrder
 import com.google.android.fhir.testing.assertResourceEquals
 import com.google.android.fhir.testing.readFromFile
@@ -44,7 +46,6 @@ import com.google.common.truth.Truth.assertThat
 import java.math.BigDecimal
 import java.time.Instant
 import java.util.Date
-import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.runBlocking
 import org.hl7.fhir.r4.model.Address
 import org.hl7.fhir.r4.model.CarePlan
@@ -504,16 +505,18 @@ class DatabaseImplTest {
         lastUpdated = Date()
       }
     database.insert(patient)
-    services.fhirEngine.syncUpload { it ->
+    val uploadMode = UploadMode.ALL_CHANGES_SQUASHED_BUNDLE_PUT(500)
+
+    services.fhirEngine.syncUpload(uploadMode) { it, resourceConsolidator ->
       it
         .first { it.resourceId == "remote-patient-3" }
         .let {
-          flowOf(
-            it.token to
-              Patient().apply {
-                id = it.resourceId
-                meta = remoteMeta
-              }
+          resourceConsolidator.consolidate(
+            Patient().apply {
+              id = it.resourceId
+              meta = remoteMeta
+            },
+            UrlUploadRequest(patient, it.token)
           )
         }
     }

engine/src/main/java/com/google/android/fhir/FhirEngine.kt

@@ -17,9 +17,10 @@
 package com.google.android.fhir
 
 import com.google.android.fhir.db.ResourceNotFoundException
-import com.google.android.fhir.db.impl.dao.LocalChangeToken
 import com.google.android.fhir.search.Search
 import com.google.android.fhir.sync.ConflictResolver
+import com.google.android.fhir.sync.upload.ResourceConsolidator
+import com.google.android.fhir.sync.upload.UploadMode
 import java.time.OffsetDateTime
 import kotlinx.coroutines.flow.Flow
 import org.hl7.fhir.r4.model.Resource
@@ -54,7 +55,8 @@ interface FhirEngine {
    * api caller should [Flow.collect] it.
    */
   suspend fun syncUpload(
-    upload: (suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>)
+    uploadMode: UploadMode,
+    upload: suspend (List<LocalChange>, ResourceConsolidator) -> Unit
   )
 
   /**

engine/src/main/java/com/google/android/fhir/impl/FhirEngineImpl.kt

@@ -21,20 +21,20 @@ import com.google.android.fhir.DatastoreUtil
 import com.google.android.fhir.FhirEngine
 import com.google.android.fhir.LocalChange
 import com.google.android.fhir.db.Database
-import com.google.android.fhir.db.impl.dao.LocalChangeToken
 import com.google.android.fhir.logicalId
 import com.google.android.fhir.search.Search
 import com.google.android.fhir.search.count
 import com.google.android.fhir.search.execute
 import com.google.android.fhir.sync.ConflictResolver
 import com.google.android.fhir.sync.Resolved
+import com.google.android.fhir.sync.upload.DefaultResourceConsolidator
+import com.google.android.fhir.sync.upload.ResourceConsolidator
+import com.google.android.fhir.sync.upload.ResourceConsolidatorStrategy
+import com.google.android.fhir.sync.upload.UploadMode
 import java.time.OffsetDateTime
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.collect
-import org.hl7.fhir.r4.model.Bundle
 import org.hl7.fhir.r4.model.Resource
 import org.hl7.fhir.r4.model.ResourceType
-import timber.log.Timber
 
 /** Implementation of [FhirEngine]. */
 internal class FhirEngineImpl(private val database: Database, private val context: Context) :
@@ -123,89 +123,17 @@ internal class FhirEngineImpl(private val database: Database, private val contex
       .intersect(database.getAllLocalChanges().map { it.resourceId }.toSet())
 
   override suspend fun syncUpload(
-    upload: suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>
+    uploadMode: UploadMode,
+    upload: suspend (List<LocalChange>, ResourceConsolidator) -> Unit,
   ) {
+    val resourceConsolidator =
+      when (uploadMode.resourceConsolidatorStrategy) {
+        ResourceConsolidatorStrategy.Default -> DefaultResourceConsolidator(database)
+        ResourceConsolidatorStrategy.IDUpdater -> error("Not yet implemented")
+      }
     val localChanges = database.getAllLocalChanges()
     if (localChanges.isNotEmpty()) {
-      upload(localChanges).collect {
-        database.deleteUpdates(it.first)
-        when (it.second) {
-          is Bundle -> updateVersionIdAndLastUpdated(it.second as Bundle)
-          else -> updateVersionIdAndLastUpdated(it.second)
-        }
-      }
+      upload(localChanges, resourceConsolidator)
     }
   }
-
-  private suspend fun updateVersionIdAndLastUpdated(bundle: Bundle) {
-    when (bundle.type) {
-      Bundle.BundleType.TRANSACTIONRESPONSE -> {
-        bundle.entry.forEach {
-          when {
-            it.hasResource() -> updateVersionIdAndLastUpdated(it.resource)
-            it.hasResponse() -> updateVersionIdAndLastUpdated(it.response)
-          }
-        }
-      }
-      else -> {
-        // Leave it for now.
-        Timber.i("Received request to update meta values for ${bundle.type}")
-      }
-    }
-  }
-
-  private suspend fun updateVersionIdAndLastUpdated(response: Bundle.BundleEntryResponseComponent) {
-    if (response.hasEtag() && response.hasLastModified() && response.hasLocation()) {
-      response.resourceIdAndType?.let { (id, type) ->
-        database.updateVersionIdAndLastUpdated(
-          id,
-          type,
-          getVersionFromETag(response.etag),
-          response.lastModified.toInstant()
-        )
-      }
-    }
-  }
-
-  private suspend fun updateVersionIdAndLastUpdated(resource: Resource) {
-    if (resource.hasMeta() && resource.meta.hasVersionId() && resource.meta.hasLastUpdated()) {
-      database.updateVersionIdAndLastUpdated(
-        resource.id,
-        resource.resourceType,
-        resource.meta.versionId,
-        resource.meta.lastUpdated.toInstant()
-      )
-    }
-  }
-
-  /**
-   * FHIR uses weak ETag that look something like W/"MTY4NDMyODE2OTg3NDUyNTAwMA", so we need to
-   * extract version from it. See https://hl7.org/fhir/http.html#Http-Headers.
-   */
-  private fun getVersionFromETag(eTag: String) =
-    // The server should always return a weak etag that starts with W, but if it server returns a
-    // strong tag, we store it as-is. The http-headers for conditional upload like if-match will
-    // always add value as a weak tag.
-    if (eTag.startsWith("W/")) {
-      eTag.split("\"")[1]
-    } else {
-      eTag
-    }
-
-  /**
-   * May return a Pair of versionId and resource type extracted from the
-   * [Bundle.BundleEntryResponseComponent.location].
-   *
-   * [Bundle.BundleEntryResponseComponent.location] may be:
-   *
-   * 1. absolute path: `<server-path>/<resource-type>/<resource-id>/_history/<version>`
-   *
-   * 2. relative path: `<resource-type>/<resource-id>/_history/<version>`
-   */
-  private val Bundle.BundleEntryResponseComponent.resourceIdAndType: Pair<String, ResourceType>?
-    get() =
-      location
-        ?.split("/")
-        ?.takeIf { it.size > 3 }
-        ?.let { it[it.size - 3] to ResourceType.fromCode(it[it.size - 4]) }
 }

engine/src/main/java/com/google/android/fhir/sync/FhirSynchronizer.kt

@@ -19,9 +19,9 @@ package com.google.android.fhir.sync
 import android.content.Context
 import com.google.android.fhir.DatastoreUtil
 import com.google.android.fhir.FhirEngine
+import com.google.android.fhir.sync.upload.UploadMode
 import java.time.OffsetDateTime
 import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flow
 import org.hl7.fhir.r4.model.ResourceType
 
@@ -125,20 +125,17 @@ internal class FhirSynchronizer(
 
   private suspend fun upload(): SyncResult {
     val exceptions = mutableListOf<ResourceSyncException>()
-    fhirEngine.syncUpload { list ->
-      flow {
-        uploader.upload(list).collect { result ->
-          when (result) {
-            is UploadState.Started ->
-              setSyncState(SyncJobStatus.InProgress(SyncOperation.UPLOAD, result.total))
-            is UploadState.Success ->
-              emit(result.localChangeToken to result.resource).also {
-                setSyncState(
-                  SyncJobStatus.InProgress(SyncOperation.UPLOAD, result.total, result.completed)
-                )
-              }
-            is UploadState.Failure -> exceptions.add(result.syncError)
-          }
+    val uploadMode = UploadMode.ALL_CHANGES_SQUASHED_BUNDLE_PUT(500)
+    fhirEngine.syncUpload(uploadMode) { list, resourceConsolidator ->
+      uploader.upload(list, resourceConsolidator).collect { result ->
+        when (result) {
+          is UploadState.Started ->
+            setSyncState(SyncJobStatus.InProgress(SyncOperation.UPLOAD, result.total))
+          is UploadState.Success ->
+            setSyncState(
+              SyncJobStatus.InProgress(SyncOperation.UPLOAD, result.total, result.completed)
+            )
+          is UploadState.Failure -> exceptions.add(result.syncError)
         }
       }
     }

engine/src/main/java/com/google/android/fhir/sync/Uploader.kt

@@ -18,6 +18,7 @@ package com.google.android.fhir.sync
 
 import com.google.android.fhir.LocalChange
 import com.google.android.fhir.db.impl.dao.LocalChangeToken
+import com.google.android.fhir.sync.upload.ResourceConsolidator
 import kotlinx.coroutines.flow.Flow
 import org.hl7.fhir.r4.model.Resource
 
@@ -29,7 +30,10 @@ internal interface Uploader {
    * transforming the [LocalChange]s to particular network operations. If [ProgressCallback] is
    * provided it also reports the intermediate progress
    */
-  suspend fun upload(localChanges: List<LocalChange>): Flow<UploadState>
+  suspend fun upload(
+    localChanges: List<LocalChange>,
+    resourceConsolidator: ResourceConsolidator
+  ): Flow<UploadState>
 }
 
 internal sealed class UploadState {

engine/src/main/java/com/google/android/fhir/sync/upload/ResourceConsolidator.kt

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.android.fhir.sync.upload
+
+import com.google.android.fhir.db.Database
+import com.google.android.fhir.sync.UploadRequest
+import org.hl7.fhir.r4.model.Bundle
+import org.hl7.fhir.r4.model.Resource
+import org.hl7.fhir.r4.model.ResourceType
+import timber.log.Timber
+
+fun interface ResourceConsolidator {
+  suspend fun consolidate(response: Resource, uploadRequest: UploadRequest)
+}
+
+internal class DefaultResourceConsolidator(private val database: Database) : ResourceConsolidator {
+
+  override suspend fun consolidate(response: Resource, uploadRequest: UploadRequest) {
+    database.deleteUpdates(uploadRequest.localChangeToken)
+    when (response) {
+      is Bundle -> updateVersionIdAndLastUpdated(response)
+      else -> updateVersionIdAndLastUpdated(response)
+    }
+  }
+
+  private suspend fun updateVersionIdAndLastUpdated(bundle: Bundle) {
+    when (bundle.type) {
+      Bundle.BundleType.TRANSACTIONRESPONSE -> {
+        bundle.entry.forEach {
+          when {
+            it.hasResource() -> updateVersionIdAndLastUpdated(it.resource)
+            it.hasResponse() -> updateVersionIdAndLastUpdated(it.response)
+          }
+        }
+      }
+      else -> {
+        // Leave it for now.
+        Timber.i("Received request to update meta values for ${bundle.type}")
+      }
+    }
+  }
+
+  private suspend fun updateVersionIdAndLastUpdated(response: Bundle.BundleEntryResponseComponent) {
+    if (response.hasEtag() && response.hasLastModified() && response.hasLocation()) {
+      response.resourceIdAndType?.let { (id, type) ->
+        database.updateVersionIdAndLastUpdated(
+          id,
+          type,
+          getVersionFromETag(response.etag),
+          response.lastModified.toInstant()
+        )
+      }
+    }
+  }
+
+  private suspend fun updateVersionIdAndLastUpdated(resource: Resource) {
+    if (resource.hasMeta() && resource.meta.hasVersionId() && resource.meta.hasLastUpdated()) {
+      database.updateVersionIdAndLastUpdated(
+        resource.id,
+        resource.resourceType,
+        resource.meta.versionId,
+        resource.meta.lastUpdated.toInstant()
+      )
+    }
+  }
+
+  /**
+   * FHIR uses weak ETag that look something like W/"MTY4NDMyODE2OTg3NDUyNTAwMA", so we need to
+   * extract version from it. See https://hl7.org/fhir/http.html#Http-Headers.
+   */
+  private fun getVersionFromETag(eTag: String) =
+    // The server should always return a weak etag that starts with W, but if it server returns a
+    // strong tag, we store it as-is. The http-headers for conditional upload like if-match will
+    // always add value as a weak tag.
+    if (eTag.startsWith("W/")) {
+      eTag.split("\"")[1]
+    } else {
+      eTag
+    }
+
+  /**
+   * May return a Pair of versionId and resource type extracted from the
+   * [Bundle.BundleEntryResponseComponent.location].
+   *
+   * [Bundle.BundleEntryResponseComponent.location] may be:
+   *
+   * 1. absolute path: `<server-path>/<resource-type>/<resource-id>/_history/<version>`
+   *
+   * 2. relative path: `<resource-type>/<resource-id>/_history/<version>`
+   */
+  private val Bundle.BundleEntryResponseComponent.resourceIdAndType: Pair<String, ResourceType>?
+    get() =
+      location
+        ?.split("/")
+        ?.takeIf { it.size > 3 }
+        ?.let { it[it.size - 3] to ResourceType.fromCode(it[it.size - 4]) }
+}