Add a LocalChangeFetcher

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.FetchMode
 import com.google.android.fhir.testing.assertJsonArrayEqualsIgnoringOrder
 import com.google.android.fhir.testing.assertResourceEquals
 import com.google.android.fhir.testing.readFromFile
@@ -512,7 +513,8 @@ class DatabaseImplTest {
         lastUpdated = Date()
       }
     database.insert(patient)
-    services.fhirEngine.syncUpload { it ->
+    services.fhirEngine.syncUpload(FetchMode.AllChanges(100)) {
+      println(it.first())
       it
         .first { it.resourceId == "remote-patient-3" }
         .let {
@@ -2370,7 +2372,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.FetchMode
 import java.time.OffsetDateTime
 import kotlinx.coroutines.flow.Flow
 import org.hl7.fhir.r4.model.Resource
@@ -54,6 +55,7 @@ interface FhirEngine {
    * api caller should [Flow.collect] it.
    */
   suspend fun syncUpload(
+    fetchMode: FetchMode,
     upload: (suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>)
   )
 

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

@@ -29,9 +29,10 @@ 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.FetchMode
+import com.google.android.fhir.sync.upload.LocalChangeFetcher
 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
@@ -124,16 +125,15 @@ internal class FhirEngineImpl(private val database: Database, private val contex
       .intersect(database.getAllLocalChanges().map { it.resourceId }.toSet())
 
   override suspend fun syncUpload(
+    fetchMode: FetchMode,
     upload: suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>
   ) {
-    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)
-        }
+    val localChangeFetcher = LocalChangeFetcher.byMode(fetchMode, database)
+    upload(localChangeFetcher.next()).collect {
+      database.deleteUpdates(it.first)
+      when (it.second) {
+        is Bundle -> updateVersionIdAndLastUpdated(it.second as Bundle)
+        else -> updateVersionIdAndLastUpdated(it.second)
       }
     }
   }

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.FetchMode
 import com.google.android.fhir.sync.upload.UploadState
 import com.google.android.fhir.sync.upload.Uploader
 import java.time.OffsetDateTime
@@ -126,7 +127,8 @@ internal class FhirSynchronizer(
 
   private suspend fun upload(): SyncResult {
     val exceptions = mutableListOf<ResourceSyncException>()
-    fhirEngine.syncUpload { list ->
+    val fetchMode = FetchMode.AllChanges(100)
+    fhirEngine.syncUpload(fetchMode) { list ->
       flow {
         uploader.upload(list).collect { result ->
           when (result) {

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

@@ -0,0 +1,58 @@
+/*
+ * 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.LocalChange
+import com.google.android.fhir.db.Database
+
+/**
+ * Defines the contract for fetching local changes.
+ *
+ * This interface provides methods to check for the existence of further changes, retrieve the next
+ * batch of changes, and get the progress of fetched changes.
+ *
+ * It is marked as internal to keep [Database] unexposed to clients
+ */
+interface LocalChangeFetcher {
+  suspend fun hasNext(): Boolean
+  suspend fun next(): List<LocalChange>
+  suspend fun getProgress(): Double
+
+  companion object {
+    internal fun byMode(mode: FetchMode, database: Database): LocalChangeFetcher =
+      when (mode) {
+        is FetchMode.AllChanges -> AllChangesLocalChangeFetcher(database)
+        else -> error("$mode does not have an implementation yet.")
+      }
+  }
+}
+
+internal class AllChangesLocalChangeFetcher(val database: Database) : LocalChangeFetcher {
+  override suspend fun hasNext(): Boolean = database.getAllLocalChanges().isNotEmpty()
+
+  override suspend fun next(): List<LocalChange> = database.getAllLocalChanges()
+
+  override suspend fun getProgress(): Double = database.getAllLocalChanges().size.toDouble()
+}
+
+/** Represents the mode in which local changes should be fetched. */
+sealed class FetchMode {
+
+  class AllChanges(val pageSize: Int) : FetchMode()
+  object PerResource : FetchMode()
+  object EarliestChange : FetchMode()
+}

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

@@ -44,7 +44,7 @@ internal class UploaderImpl(
     val transformedChanges = uploadWorkManager.prepareChangesForUpload(localChanges)
     val uploadRequests = uploadWorkManager.createUploadRequestsFromLocalChanges(transformedChanges)
     val total = uploadWorkManager.getPendingUploadsIndicator(uploadRequests)
-    var completed = 0
+    var completed: Int
     emit(UploadState.Started(total))
     val pendingRequests = uploadRequests.toMutableList()
     while (pendingRequests.isNotEmpty()) {

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.FetchMode
 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(
+    fetchMode: FetchMode,
     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.FetchMode
 import com.google.android.fhir.testing.assertResourceEquals
 import com.google.android.fhir.testing.assertResourceNotEquals
 import com.google.android.fhir.testing.readFromFile
@@ -313,15 +314,14 @@ class FhirEngineImplTest {
   @Test
   fun syncUpload_uploadLocalChange() = runBlocking {
     val localChanges = mutableListOf<LocalChange>()
-    fhirEngine.syncUpload {
+    fhirEngine.syncUpload(FetchMode.AllChanges(100)) {
       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 +414,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()