Adding PerResourceLocalChangeFetcher

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

@@ -233,6 +233,23 @@ class DatabaseImplTest {
     assertThat(database.getLocalChanges(ResourceType.Encounter, patient.logicalId)).isEmpty()
   }
 
+  @Test
+  fun getAllChangesForEarliestChangedResource_withMultipleChanges_shouldReturnFirstChange() =
+    runBlocking {
+      val patient: Patient = readFromFile(Patient::class.java, "/date_test_patient.json")
+      database.insert(patient)
+      database.insert(TEST_PATIENT_2)
+      database.update(
+        TEST_PATIENT_1.copy().apply { gender = Enumerations.AdministrativeGender.FEMALE },
+      )
+      assertThat(
+          database.getAllChangesForEarliestChangedResource().all {
+            it.resourceId.equals(TEST_PATIENT_1.logicalId)
+          },
+        )
+        .isTrue()
+    }
+
   @Test
   fun clearDatabase_shouldClearAllTablesData() = runBlocking {
     val patient: Patient = readFromFile(Patient::class.java, "/date_test_patient.json")

engine/src/main/java/com/google/android/fhir/db/Database.kt

@@ -105,6 +105,12 @@ internal interface Database {
    */
   suspend fun getAllLocalChanges(): List<LocalChange>
 
+  /**
+   * Retrieves all [LocalChange]s for the [Resource] which has the [LocalChange] with the oldest
+   * [LocalChange.timestamp]
+   */
+  suspend fun getAllChangesForEarliestChangedResource(): List<LocalChange>
+
   /** Retrieves the count of [LocalChange]s stored in the database. */
   suspend fun getLocalChangesCount(): Int
 

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

@@ -228,6 +228,10 @@ internal class DatabaseImpl(
     return db.withTransaction { localChangeDao.getLocalChangesCount() }
   }
 
+  override suspend fun getAllChangesForEarliestChangedResource(): List<LocalChange> {
+    return localChangeDao.getAllChangesForEarliestChangedResource().map { it.toLocalChange() }
+  }
+
   override suspend fun deleteUpdates(token: LocalChangeToken) {
     db.withTransaction { localChangeDao.discardLocalChanges(token) }
   }

engine/src/main/java/com/google/android/fhir/db/impl/dao/LocalChangeDao.kt

@@ -206,6 +206,20 @@ internal abstract class LocalChangeDao {
     resourceId: String,
   ): List<LocalChangeEntity>
 
+  @Query(
+    """
+      SELECT * 
+      FROM LocalChangeEntity 
+      WHERE resourceUuid = (
+        SELECT resourceUuid 
+        FROM LocalChangeEntity 
+        ORDER BY timestamp ASC 
+        LIMIT 1)
+      ORDER BY timestamp ASC
+    """,
+  )
+  abstract suspend fun getAllChangesForEarliestChangedResource(): List<LocalChangeEntity>
+
   class InvalidLocalChangeException(message: String?) : Exception(message)
 }
 

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

@@ -71,6 +71,25 @@ internal class AllChangesLocalChangeFetcher(
     SyncUploadProgress(database.getLocalChangesCount(), total)
 }
 
+internal class PerResourceLocalChangeFetcher(
+  private val database: Database,
+) : LocalChangeFetcher {
+
+  override var total by Delegates.notNull<Int>()
+
+  suspend fun initTotalCount() {
+    total = database.getLocalChangesCount()
+  }
+
+  override suspend fun hasNext(): Boolean = database.getLocalChangesCount().isNotZero()
+
+  override suspend fun next(): List<LocalChange> =
+    database.getAllChangesForEarliestChangedResource()
+
+  override suspend fun getProgress(): SyncUploadProgress =
+    SyncUploadProgress(database.getLocalChangesCount(), total)
+}
+
 /** Represents the mode in which local changes should be fetched. */
 sealed class LocalChangesFetchMode {
   object AllChanges : LocalChangesFetchMode()
@@ -88,6 +107,8 @@ internal object LocalChangeFetcherFactory {
     when (mode) {
       is LocalChangesFetchMode.AllChanges ->
         AllChangesLocalChangeFetcher(database).apply { initTotalCount() }
+      is LocalChangesFetchMode.PerResource ->
+        PerResourceLocalChangeFetcher(database).apply { initTotalCount() }
       else -> throw NotImplementedError("$mode is not implemented yet.")
     }
 }

engine/src/test/java/com/google/android/fhir/sync/upload/AllChangesLocalChangeFetcherTest.kt → engine/src/test/java/com/google/android/fhir/sync/upload/fetcher/AllChangesLocalChangeFetcherTest.kt

@@ -14,10 +14,12 @@
  * limitations under the License.
  */
 
-package com.google.android.fhir.sync.upload
+package com.google.android.fhir.sync.upload.fetcher
 
 import androidx.test.core.app.ApplicationProvider
 import com.google.android.fhir.FhirServices
+import com.google.android.fhir.sync.upload.AllChangesLocalChangeFetcher
+import com.google.android.fhir.sync.upload.SyncUploadProgress
 import com.google.common.truth.Truth.assertThat
 import java.util.Date
 import kotlinx.coroutines.test.runTest

engine/src/test/java/com/google/android/fhir/sync/upload/fetcher/PerResourceLocalChangeFetcherTest.kt

@@ -0,0 +1,114 @@
+/*
+ * 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.fetcher
+
+import androidx.test.core.app.ApplicationProvider
+import com.google.android.fhir.FhirServices
+import com.google.android.fhir.LocalChange
+import com.google.android.fhir.logicalId
+import com.google.android.fhir.sync.upload.PerResourceLocalChangeFetcher
+import com.google.common.truth.Truth.assertThat
+import java.util.Date
+import kotlinx.coroutines.test.runTest
+import org.hl7.fhir.r4.model.Enumerations
+import org.hl7.fhir.r4.model.Meta
+import org.hl7.fhir.r4.model.Patient
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+
+@RunWith(RobolectricTestRunner::class)
+class PerResourceLocalChangeFetcherTest {
+
+  private val services =
+    FhirServices.builder(ApplicationProvider.getApplicationContext()).inMemory().build()
+  private val database = services.database
+
+  @Test
+  fun `fetcher is created correctly`() = runTest {
+    database.insert(TEST_PATIENT_1, TEST_PATIENT_2)
+    database.update(
+      TEST_PATIENT_1.copy().apply { gender = Enumerations.AdministrativeGender.FEMALE },
+    )
+    val fetcher = PerResourceLocalChangeFetcher(database).apply { initTotalCount() }
+
+    assertThat(fetcher.getProgress().initialTotal).isEqualTo(3)
+  }
+
+  @Test
+  fun `hasNext returns correct value`() = runTest {
+    database.insert(TEST_PATIENT_1, TEST_PATIENT_2)
+    database.update(
+      TEST_PATIENT_1.copy().apply { gender = Enumerations.AdministrativeGender.FEMALE },
+    )
+    val fetcher = PerResourceLocalChangeFetcher(database).apply { initTotalCount() }
+
+    assertThat(fetcher.hasNext()).isTrue()
+    database.deleteUpdates(listOf(TEST_PATIENT_1))
+    assertThat(fetcher.hasNext()).isTrue()
+    database.deleteUpdates(listOf(TEST_PATIENT_2))
+    assertThat(fetcher.hasNext()).isFalse()
+  }
+
+  @Test
+  fun `next returns correct set of changes in the right order`() = runTest {
+    database.insert(TEST_PATIENT_1, TEST_PATIENT_2)
+    database.update(
+      TEST_PATIENT_1.copy().apply { gender = Enumerations.AdministrativeGender.FEMALE },
+    )
+    val fetcher = PerResourceLocalChangeFetcher(database).apply { initTotalCount() }
+
+    val firstSetOfChanges = fetcher.next()
+    database.deleteUpdates(listOf(TEST_PATIENT_1))
+    val secondSetOfChanges = fetcher.next()
+    database.deleteUpdates(listOf(TEST_PATIENT_2))
+
+    assertThat(firstSetOfChanges.size).isEqualTo(2)
+    with(firstSetOfChanges[0]) {
+      assertThat(type).isEqualTo(LocalChange.Type.INSERT)
+      assertThat(resourceId).isEqualTo(TEST_PATIENT_1.logicalId)
+    }
+
+    with(firstSetOfChanges[1]) {
+      assertThat(type).isEqualTo(LocalChange.Type.UPDATE)
+      assertThat(resourceId).isEqualTo(TEST_PATIENT_1.logicalId)
+    }
+
+    assertThat(secondSetOfChanges.size).isEqualTo(1)
+    with(secondSetOfChanges[0]) {
+      assertThat(type).isEqualTo(LocalChange.Type.INSERT)
+      assertThat(resourceId).isEqualTo(TEST_PATIENT_2.logicalId)
+    }
+  }
+
+  companion object {
+    private const val TEST_PATIENT_1_ID = "test_patient_1"
+    private var TEST_PATIENT_1 =
+      Patient().apply {
+        id = TEST_PATIENT_1_ID
+        gender = Enumerations.AdministrativeGender.MALE
+      }
+
+    private const val TEST_PATIENT_2_ID = "test_patient_2"
+    private var TEST_PATIENT_2 =
+      Patient().apply {
+        id = TEST_PATIENT_2_ID
+        gender = Enumerations.AdministrativeGender.MALE
+        meta = Meta().apply { lastUpdated = Date() }
+      }
+  }
+}