Add ResourceConsolidator

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

@@ -39,6 +39,7 @@ import com.google.android.fhir.search.getQuery
 import com.google.android.fhir.search.has
 import com.google.android.fhir.search.include
 import com.google.android.fhir.search.revInclude
+import com.google.android.fhir.sync.upload.ConsolidatorMode
 import com.google.android.fhir.testing.assertJsonArrayEqualsIgnoringOrder
 import com.google.android.fhir.testing.assertResourceEquals
 import com.google.android.fhir.testing.readFromFile
@@ -512,7 +513,7 @@ class DatabaseImplTest {
         lastUpdated = Date()
       }
     database.insert(patient)
-    services.fhirEngine.syncUpload { it ->
+    services.fhirEngine.syncUpload(ConsolidatorMode.Default) { it ->
       it
         .first { it.resourceId == "remote-patient-3" }
         .let {
@@ -2370,7 +2371,7 @@ class DatabaseImplTest {
 
   @Test
   fun search_nameGivenDuplicate_deduplicatePatient() = runBlocking {
-    var patient: Patient = readFromFile(Patient::class.java, "/patient_name_given_duplicate.json")
+    val patient: Patient = readFromFile(Patient::class.java, "/patient_name_given_duplicate.json")
     database.insertRemote(patient)
     val result =
       database.search<Patient>(

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

@@ -20,6 +20,7 @@ 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.ConsolidatorMode
 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>>)
+    consolidatorMode: ConsolidatorMode,
+    upload: suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>
   )
 
   /**

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

@@ -29,13 +29,12 @@ 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.ConsolidatorMode
+import com.google.android.fhir.sync.upload.ResourceConsolidator
 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) :
@@ -124,89 +123,13 @@ 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>>
+    consolidatorMode: ConsolidatorMode,
+    upload: suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>,
   ) {
+    val resourceConsolidator = ResourceConsolidator.byMode(consolidatorMode, database)
     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)
-        }
-      }
-    }
-  }
-
-  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()
-      )
+      upload(localChanges).collect { resourceConsolidator.consolidate(it.first, it.second) }
     }
   }
-
-  /**
-   * 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,6 +19,7 @@ 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.ConsolidatorMode
 import com.google.android.fhir.sync.upload.UploadState
 import com.google.android.fhir.sync.upload.Uploader
 import java.time.OffsetDateTime
@@ -126,7 +127,7 @@ internal class FhirSynchronizer(
 
   private suspend fun upload(): SyncResult {
     val exceptions = mutableListOf<ResourceSyncException>()
-    fhirEngine.syncUpload { list ->
+    fhirEngine.syncUpload(ConsolidatorMode.Default) { list ->
       flow {
         uploader.upload(list).collect { result ->
           when (result) {

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

@@ -0,0 +1,123 @@
+/*
+ * 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.db.impl.dao.LocalChangeToken
+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(localChangeToken: LocalChangeToken, response: Resource)
+  companion object {
+    internal fun byMode(mode: ConsolidatorMode, database: Database): ResourceConsolidator =
+      when (mode) {
+        is ConsolidatorMode.Default -> DefaultResourceConsolidator(database)
+        else -> error("$mode does not have an implementation yet.")
+      }
+  }
+}
+
+internal class DefaultResourceConsolidator(private val database: Database) : ResourceConsolidator {
+
+  override suspend fun consolidate(localChangeToken: LocalChangeToken, response: Resource) {
+    database.deleteUpdates(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]) }
+}
+
+sealed class ConsolidatorMode {
+  object IDUpdater : ConsolidatorMode()
+  object Default : ConsolidatorMode()
+}

engine/src/main/java/com/google/android/fhir/testing/Utilities.kt

@@ -33,6 +33,7 @@ import com.google.android.fhir.sync.DownloadRequest
 import com.google.android.fhir.sync.DownloadWorkManager
 import com.google.android.fhir.sync.UploadRequest
 import com.google.android.fhir.sync.UrlDownloadRequest
+import com.google.android.fhir.sync.upload.ConsolidatorMode
 import com.google.common.truth.Truth.assertThat
 import java.net.SocketTimeoutException
 import java.time.Instant
@@ -147,10 +148,9 @@ object TestFhirEngineImpl : FhirEngine {
   }
 
   override suspend fun syncUpload(
+    consolidatorMode: ConsolidatorMode,
     upload: suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>
-  ) {
-    upload(getLocalChanges(ResourceType.Patient, "123")).collect()
-  }
+  ) = upload(getLocalChanges(ResourceType.Patient, "123")).collect()
 
   override suspend fun syncDownload(
     conflictResolver: ConflictResolver,

engine/src/test/java/com/google/android/fhir/impl/FhirEngineImplTest.kt

@@ -29,6 +29,7 @@ import com.google.android.fhir.search.LOCAL_LAST_UPDATED_PARAM
 import com.google.android.fhir.search.search
 import com.google.android.fhir.sync.AcceptLocalConflictResolver
 import com.google.android.fhir.sync.AcceptRemoteConflictResolver
+import com.google.android.fhir.sync.upload.ConsolidatorMode
 import com.google.android.fhir.testing.assertResourceEquals
 import com.google.android.fhir.testing.assertResourceNotEquals
 import com.google.android.fhir.testing.readFromFile
@@ -313,15 +314,13 @@ class FhirEngineImplTest {
   @Test
   fun syncUpload_uploadLocalChange() = runBlocking {
     val localChanges = mutableListOf<LocalChange>()
-    fhirEngine.syncUpload {
+    fhirEngine.syncUpload(ConsolidatorMode.Default) {
       flow {
         localChanges.addAll(it)
         emit(LocalChangeToken(it.flatMap { it.token.ids }) to TEST_PATIENT_1)
       }
     }
-
     assertThat(localChanges).hasSize(1)
-    // val localChange = localChanges[0].localChange
     with(localChanges[0]) {
       assertThat(this.resourceType).isEqualTo(ResourceType.Patient.toString())
       assertThat(this.resourceId).isEqualTo(TEST_PATIENT_1.id)
@@ -414,7 +413,7 @@ class FhirEngineImplTest {
       assertThat(get(0).payload).isEqualTo(patientString)
     }
     assertResourceEquals(patient, fhirEngine.get(ResourceType.Patient, patient.logicalId))
-    // clear databse
+    // clear database
     runBlocking(Dispatchers.IO) { fhirEngine.clearDatabase() }
     // assert that previously present resource not available after clearing database
     assertThat(fhirEngine.getLocalChanges(patient.resourceType, patient.logicalId)).isEmpty()