FhirSyncDbInteractor

demo/src/main/java/com/google/android/fhir/demo/data/DemoFhirSyncWorker.kt

@@ -34,6 +34,4 @@ class DemoFhirSyncWorker(appContext: Context, workerParams: WorkerParameters) :
   override fun getConflictResolver() = AcceptLocalConflictResolver
 
   override fun getUploadStrategy(): UploadStrategy = UploadStrategy.AllChangesSquashedBundlePut
-
-  override fun getFhirEngine() = FhirApplication.fhirEngine(applicationContext)
 }

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

@@ -18,12 +18,7 @@ package com.google.android.fhir
 
 import com.google.android.fhir.db.ResourceNotFoundException
 import com.google.android.fhir.search.Search
-import com.google.android.fhir.sync.ConflictResolver
-import com.google.android.fhir.sync.upload.LocalChangesFetchMode
-import com.google.android.fhir.sync.upload.SyncUploadProgress
-import com.google.android.fhir.sync.upload.UploadRequestResult
 import java.time.OffsetDateTime
-import kotlinx.coroutines.flow.Flow
 import org.hl7.fhir.r4.model.Resource
 import org.hl7.fhir.r4.model.ResourceType
 
@@ -50,31 +45,6 @@ interface FhirEngine {
    */
   suspend fun <R : Resource> search(search: Search): List<SearchResult<R>>
 
-  /**
-   * Synchronizes the upload results in the database.
-   *
-   * The [upload] function may initiate multiple server calls. Each call's result can then be used
-   * to emit [UploadSyncResult]. The caller should collect these results using [Flow.collect].
-   *
-   * @param localChangesFetchMode Specifies the mode to fetch local changes.
-   * @param upload A suspend function that takes a list of [LocalChange] and returns an
-   *   [UploadSyncResult].
-   * @return A [Flow] that emits the progress of the synchronization process.
-   */
-  suspend fun syncUpload(
-    localChangesFetchMode: LocalChangesFetchMode,
-    upload: (suspend (List<LocalChange>) -> Flow<UploadRequestResult>),
-  ): Flow<SyncUploadProgress>
-
-  /**
-   * Synchronizes the [download] result in the database. The database will be updated to reflect the
-   * result of the [download] operation.
-   */
-  suspend fun syncDownload(
-    conflictResolver: ConflictResolver,
-    download: suspend () -> Flow<List<Resource>>,
-  )
-
   /**
    * Returns the total count of entities available for given search.
    *

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

@@ -18,6 +18,7 @@ package com.google.android.fhir
 
 import android.content.Context
 import com.google.android.fhir.DatabaseErrorStrategy.UNSPECIFIED
+import com.google.android.fhir.db.Database
 import com.google.android.fhir.sync.DataSource
 import com.google.android.fhir.sync.FhirDataStore
 import com.google.android.fhir.sync.HttpAuthenticator
@@ -67,6 +68,12 @@ object FhirEngineProvider {
     return getOrCreateFhirService(context).fhirDataStore
   }
 
+  @PublishedApi
+  @Synchronized
+  internal fun getFhirDatabase(context: Context): Database {
+    return getOrCreateFhirService(context).database
+  }
+
   @Synchronized
   private fun getOrCreateFhirService(context: Context): FhirServices {
     if (fhirServices == null) {

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

@@ -0,0 +1,158 @@
+/*
+ * Copyright 2024 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
+
+import com.google.android.fhir.db.Database
+import com.google.android.fhir.sync.ConflictResolver
+import com.google.android.fhir.sync.Resolved
+import com.google.android.fhir.sync.SyncJobStatus
+import com.google.android.fhir.sync.SyncOperation
+import com.google.android.fhir.sync.download.DownloadState
+import com.google.android.fhir.sync.upload.LocalChangeFetcher
+import com.google.android.fhir.sync.upload.ResourceConsolidator
+import com.google.android.fhir.sync.upload.UploadRequestResult
+import org.hl7.fhir.r4.model.Resource
+
+internal interface FhirSyncDbInteractor {
+  suspend fun getLocalChanges(): List<LocalChange>
+
+  suspend fun consolidateUploadResult(uploadRequestResult: UploadRequestResult)
+
+  suspend fun consolidateDownloadResult(downloadState: DownloadState)
+
+  suspend fun updateSyncJobStatus(
+    previousSyncJobStatus: SyncJobStatus,
+    downloadState: DownloadState,
+  ): SyncJobStatus
+
+  suspend fun updateSyncJobStatus(
+    previousSyncJobStatus: SyncJobStatus,
+    uploadRequestResult: UploadRequestResult,
+  ): SyncJobStatus
+}
+
+internal class FhirSyncDbInteractorImpl(
+  private val database: Database,
+  private val localChangeFetcher: LocalChangeFetcher,
+  private val resourceConsolidator: ResourceConsolidator,
+  private val conflictResolver: ConflictResolver,
+) : FhirSyncDbInteractor {
+  override suspend fun getLocalChanges() = localChangeFetcher.next()
+
+  override suspend fun consolidateUploadResult(uploadRequestResult: UploadRequestResult) {
+    resourceConsolidator.consolidate(uploadRequestResult)
+  }
+
+  override suspend fun consolidateDownloadResult(downloadState: DownloadState) {
+    when (downloadState) {
+      is DownloadState.Started -> {
+        /**  */
+      }
+      is DownloadState.Success -> {
+        database.withTransaction {
+          val resolved =
+            resolveConflictingResources(
+              downloadState.resources,
+              getConflictingResourceIds(downloadState.resources),
+              conflictResolver,
+            )
+          database.insertSyncedResources(downloadState.resources)
+          saveResolvedResourcesToDatabase(resolved)
+        }
+      }
+      is DownloadState.Failure -> {}
+    }
+  }
+
+  override suspend fun updateSyncJobStatus(
+    previousSyncJobStatus: SyncJobStatus,
+    downloadState: DownloadState,
+  ): SyncJobStatus {
+    return with(previousSyncJobStatus as SyncJobStatus.InProgress) {
+      when (downloadState) {
+        is DownloadState.Started -> {
+          SyncJobStatus.InProgress(
+            syncOperation = SyncOperation.DOWNLOAD,
+            total = downloadState.total,
+            completed = 0,
+          )
+        }
+        is DownloadState.Success -> {
+          SyncJobStatus.InProgress(
+            syncOperation = syncOperation,
+            total = downloadState.total,
+            completed = downloadState.completed,
+          )
+        }
+        is DownloadState.Failure -> {
+          SyncJobStatus.Failed(
+            exceptions = listOf(downloadState.syncError),
+          )
+        }
+      }
+    }
+  }
+
+  override suspend fun updateSyncJobStatus(
+    previousSyncJobStatus: SyncJobStatus,
+    uploadRequestResult: UploadRequestResult,
+  ): SyncJobStatus {
+    return when (uploadRequestResult) {
+      is UploadRequestResult.Success -> {
+        val localChangesCount =
+          uploadRequestResult.successfulUploadResponseMappings.flatMap { it.localChanges }.size
+        with(previousSyncJobStatus as SyncJobStatus.InProgress) {
+          SyncJobStatus.InProgress(
+            syncOperation = syncOperation,
+            completed = completed + localChangesCount,
+            total = total,
+          )
+        }
+      }
+      is UploadRequestResult.Failure -> {
+        SyncJobStatus.Failed(
+          exceptions = listOf(uploadRequestResult.uploadError),
+        )
+      }
+    }
+  }
+
+  private suspend fun saveResolvedResourcesToDatabase(resolved: List<Resource>?) {
+    resolved?.let {
+      database.deleteUpdates(it)
+      database.update(*it.toTypedArray())
+    }
+  }
+
+  private suspend fun resolveConflictingResources(
+    resources: List<Resource>,
+    conflictingResourceIds: Set<String>,
+    conflictResolver: ConflictResolver,
+  ) =
+    resources
+      .filter { conflictingResourceIds.contains(it.logicalId) }
+      .map { conflictResolver.resolve(database.select(it.resourceType, it.logicalId), it) }
+      .filterIsInstance<Resolved>()
+      .map { it.resolved }
+      .takeIf { it.isNotEmpty() }
+
+  private suspend fun getConflictingResourceIds(resources: List<Resource>) =
+    resources
+      .map { it.logicalId }
+      .toSet()
+      .intersect(database.getAllLocalChanges().map { it.resourceId }.toSet())
+}

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

@@ -22,22 +22,10 @@ import com.google.android.fhir.FhirEngineProvider
 import com.google.android.fhir.LocalChange
 import com.google.android.fhir.SearchResult
 import com.google.android.fhir.db.Database
-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.LocalChangeFetcherFactory
-import com.google.android.fhir.sync.upload.LocalChangesFetchMode
-import com.google.android.fhir.sync.upload.SyncUploadProgress
-import com.google.android.fhir.sync.upload.UploadRequestResult
 import java.time.OffsetDateTime
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.firstOrNull
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.flow.onEach
 import org.hl7.fhir.r4.model.Resource
 import org.hl7.fhir.r4.model.ResourceType
 
@@ -83,83 +71,4 @@ internal class FhirEngineImpl(private val database: Database, private val contex
   override suspend fun purge(type: ResourceType, id: String, forcePurge: Boolean) {
     database.purge(type, id, forcePurge)
   }
-
-  override suspend fun syncDownload(
-    conflictResolver: ConflictResolver,
-    download: suspend () -> Flow<List<Resource>>,
-  ) {
-    download().collect { resources ->
-      database.withTransaction {
-        val resolved =
-          resolveConflictingResources(
-            resources,
-            getConflictingResourceIds(resources),
-            conflictResolver,
-          )
-        database.insertSyncedResources(resources)
-        saveResolvedResourcesToDatabase(resolved)
-      }
-    }
-  }
-
-  private suspend fun saveResolvedResourcesToDatabase(resolved: List<Resource>?) {
-    resolved?.let {
-      database.deleteUpdates(it)
-      database.update(*it.toTypedArray())
-    }
-  }
-
-  private suspend fun resolveConflictingResources(
-    resources: List<Resource>,
-    conflictingResourceIds: Set<String>,
-    conflictResolver: ConflictResolver,
-  ) =
-    resources
-      .filter { conflictingResourceIds.contains(it.logicalId) }
-      .map { conflictResolver.resolve(database.select(it.resourceType, it.logicalId), it) }
-      .filterIsInstance<Resolved>()
-      .map { it.resolved }
-      .takeIf { it.isNotEmpty() }
-
-  private suspend fun getConflictingResourceIds(resources: List<Resource>) =
-    resources
-      .map { it.logicalId }
-      .toSet()
-      .intersect(database.getAllLocalChanges().map { it.resourceId }.toSet())
-
-  override suspend fun syncUpload(
-    localChangesFetchMode: LocalChangesFetchMode,
-    upload: (suspend (List<LocalChange>) -> Flow<UploadRequestResult>),
-  ): Flow<SyncUploadProgress> = flow {
-    val resourceConsolidator = DefaultResourceConsolidator(database)
-    val localChangeFetcher = LocalChangeFetcherFactory.byMode(localChangesFetchMode, database)
-
-    emit(
-      SyncUploadProgress(
-        remaining = localChangeFetcher.total,
-        initialTotal = localChangeFetcher.total,
-      ),
-    )
-
-    while (localChangeFetcher.hasNext()) {
-      val localChanges = localChangeFetcher.next()
-      val uploadRequestResult =
-        upload(localChanges)
-          .onEach { result ->
-            resourceConsolidator.consolidate(result)
-            val newProgress =
-              when (result) {
-                is UploadRequestResult.Success -> localChangeFetcher.getProgress()
-                is UploadRequestResult.Failure ->
-                  localChangeFetcher.getProgress().copy(uploadError = result.uploadError)
-              }
-            emit(newProgress)
-          }
-          .firstOrNull { it is UploadRequestResult.Failure }
-
-      if (uploadRequestResult is UploadRequestResult.Failure) {
-        break
-      }
-    }
-  }
 }