Add a LocalChangeFetcher (#2135)

* DRAFT: add a localchangeselector * just have one class * move iterations to syncUpload * address comments * add tests for allchangeslocalchangefetcher * add progress state * spotless * add count api to db and refactor mode init * try change * spotless * Update engine/src/main/java/com/google/android/fhir/sync/upload/LocalChangeFetcher.kt Co-authored-by: Jing Tang <[email protected]> --------- Co-authored-by: Jing Tang <[email protected]>
google · Sep 14, 2023 · d639b63 · d639b63
1 parent 0bbdeb9
commit d639b63
Show file tree

Hide file tree

Showing 11 changed files with 620 additions and 360 deletions.
diff --git a/engine/src/androidTest/java/com/google/android/fhir/db/impl/DatabaseImplTest.kt b/engine/src/androidTest/java/com/google/android/fhir/db/impl/DatabaseImplTest.kt
diff --git a/engine/src/main/java/com/google/android/fhir/FhirEngine.kt b/engine/src/main/java/com/google/android/fhir/FhirEngine.kt
@@ -19,6 +19,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 java.time.OffsetDateTime
 import kotlinx.coroutines.flow.Flow
 import org.hl7.fhir.r4.model.Resource
@@ -53,7 +54,8 @@ interface FhirEngine {
  * api caller should [Flow.collect] it.
  */
  suspend fun syncUpload(
- upload: (suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>)
+ localChangesFetchMode: LocalChangesFetchMode,
+ upload: (suspend (List<LocalChange>) -> Flow<Pair<LocalChangeToken, Resource>>),
  )
 
  /**
@@ -62,7 +64,7 @@ interface FhirEngine {
  */
  suspend fun syncDownload(
  conflictResolver: ConflictResolver,
- download: suspend () -> Flow<List<Resource>>
+ download: suspend () -> Flow<List<Resource>>,
  )
 
  /**
@@ -87,29 +89,32 @@ interface FhirEngine {
  * Retrieves a list of [LocalChange]s for [Resource] with given type and id, which can be used to
  * purge resource from database. If there is no local change for given [resourceType] and
  * [Resource.id], return an empty list.
+ *
  * @param type The [ResourceType]
  * @param id The resource id [Resource.id]
  * @return [List]<[LocalChange]> A list of local changes for given [resourceType] and
- * [Resource.id] . If there is no local change for given [resourceType] and [Resource.id], return
- * an empty list.
+ *  [Resource.id] . If there is no local change for given [resourceType] and [Resource.id],
+ *  return an empty list.
  */
  suspend fun getLocalChanges(type: ResourceType, id: String): List<LocalChange>
 
  /**
  * Purges a resource from the database based on resource type and id without any deletion of data
  * from the server.
+ *
  * @param type The [ResourceType]
  * @param id The resource id [Resource.id]
  * @param isLocalPurge default value is false here resource will not be deleted from
- * LocalChangeEntity table but it will throw IllegalStateException("Resource has local changes
- * either sync with server or FORCE_PURGE required") if local change exists. If true this API will
- * delete resource entry from LocalChangeEntity table.
+ *  LocalChangeEntity table but it will throw IllegalStateException("Resource has local changes
+ *  either sync with server or FORCE_PURGE required") if local change exists. If true this API
+ *  will delete resource entry from LocalChangeEntity table.
  */
  suspend fun purge(type: ResourceType, id: String, forcePurge: Boolean = false)
 }
 
 /**
  * Returns a FHIR resource of type [R] with [id] from the local storage.
+ *
  * @param <R> The resource type which should be a subtype of [Resource].
  * @throws ResourceNotFoundException if the resource is not found
  */
@@ -120,6 +125,7 @@ suspend inline fun <reified R : Resource> FhirEngine.get(id: String): R {
 
 /**
  * Deletes a FHIR resource of type [R] with [id] from the local storage.
+ *
  * @param <R> The resource type which should be a subtype of [Resource].
  */
 suspend inline fun <reified R : Resource> FhirEngine.delete(id: String) {
@@ -138,7 +144,7 @@ data class SearchResult<R : Resource>(
  /** Matching referenced resources as per the [Search.include] criteria in the query. */
  val included: Map<SearchParamName, List<Resource>>?,
  /** Matching referenced resources as per the [Search.revInclude] criteria in the query. */
- val revIncluded: Map<Pair<ResourceType, SearchParamName>, List<Resource>>?
+ val revIncluded: Map<Pair<ResourceType, SearchParamName>, List<Resource>>?,
 ) {
  override fun equals(other: Any?) =
  other is SearchResult<*> &&
@@ -155,7 +161,7 @@ data class SearchResult<R : Resource>(
 
  private fun equalsShallow(
  first: Map<SearchParamName, List<Resource>>?,
- second: Map<SearchParamName, List<Resource>>?
+ second: Map<SearchParamName, List<Resource>>?,
  ) =
  if (first != null && second != null && first.size == second.size) {
  first.entries.asSequence().zip(second.entries.asSequence()).all { (x, y) ->
@@ -168,7 +174,7 @@ data class SearchResult<R : Resource>(
  @JvmName("equalsShallowRevInclude")
  private fun equalsShallow(
  first: Map<Pair<ResourceType, SearchParamName>, List<Resource>>?,
- second: Map<Pair<ResourceType, SearchParamName>, List<Resource>>?
+ second: Map<Pair<ResourceType, SearchParamName>, List<Resource>>?,
  ) =
  if (first != null && second != null && first.size == second.size) {
  first.entries.asSequence().zip(second.entries.asSequence()).all { (x, y) ->

diff --git a/engine/src/main/java/com/google/android/fhir/db/Database.kt b/engine/src/main/java/com/google/android/fhir/db/Database.kt
@@ -58,7 +58,7 @@ internal interface Database {
  resourceId: String,
  resourceType: ResourceType,
  versionId: String,
- lastUpdated: Instant
+ lastUpdated: Instant,
  )
 
  /**
@@ -105,6 +105,9 @@ internal interface Database {
  */
  suspend fun getAllLocalChanges(): List<LocalChange>
 
+ /** Retrieves the count of [LocalChange]s stored in the database. */
+ suspend fun getLocalChangesCount(): Int
+
  /** Remove the [LocalChangeEntity] s with given ids. Call this after a successful sync. */
  suspend fun deleteUpdates(token: LocalChangeToken)
 
@@ -127,23 +130,25 @@ internal interface Database {
  * Retrieve a list of [LocalChange] for [Resource] with given type and id, which can be used to
  * purge resource from database. If there is no local change for given [resourceType] and
  * [Resource.id], return an empty list.
+ *
  * @param type The [ResourceType]
  * @param id The resource id [Resource.id]
  * @return [List]<[LocalChange]> A list of local changes for given [resourceType] and
- * [Resource.id] . If there is no local change for given [resourceType] and [Resource.id], return
- * empty list.
+ *  [Resource.id] . If there is no local change for given [resourceType] and [Resource.id],
+ *  return empty list.
  */
  suspend fun getLocalChanges(type: ResourceType, id: String): List<LocalChange>
 
  /**
  * Purge resource from database based on resource type and id without any deletion of data from
  * the server.
+ *
  * @param type The [ResourceType]
  * @param id The resource id [Resource.id]
  * @param isLocalPurge default value is false here resource will not be deleted from
- * LocalChangeEntity table but it will throw IllegalStateException("Resource has local changes
- * either sync with server or FORCE_PURGE required") if local change exists. If true this API will
- * delete resource entry from LocalChangeEntity table.
+ *  LocalChangeEntity table but it will throw IllegalStateException("Resource has local changes
+ *  either sync with server or FORCE_PURGE required") if local change exists. If true this API
+ *  will delete resource entry from LocalChangeEntity table.
  */
  suspend fun purge(type: ResourceType, id: String, forcePurge: Boolean = false)
 }
diff --git a/engine/src/main/java/com/google/android/fhir/db/impl/DatabaseImpl.kt b/engine/src/main/java/com/google/android/fhir/db/impl/DatabaseImpl.kt
@@ -46,7 +46,7 @@ internal class DatabaseImpl(
  private val context: Context,
  private val iParser: IParser,
  databaseConfig: DatabaseConfig,
- private val resourceIndexer: ResourceIndexer
+ private val resourceIndexer: ResourceIndexer,
 ) : com.google.android.fhir.db.Database {
 
  val db: ResourceDatabase
@@ -88,8 +88,10 @@ internal class DatabaseImpl(
  openHelperFactory {
  SQLCipherSupportHelper(
  it,
- databaseErrorStrategy = databaseConfig.databaseErrorStrategy
- ) { DatabaseEncryptionKeyProvider.getOrCreatePassphrase(DATABASE_PASSPHRASE_NAME) }
+ databaseErrorStrategy = databaseConfig.databaseErrorStrategy,
+ ) {
+ DatabaseEncryptionKeyProvider.getOrCreatePassphrase(DATABASE_PASSPHRASE_NAME)
+ }
  }
  }
 
@@ -115,7 +117,7 @@ internal class DatabaseImpl(
  val timeOfLocalChange = Instant.now()
  localChangeDao.addInsert(it, timeOfLocalChange)
  resourceDao.insertLocalResource(it, timeOfLocalChange)
- }
+ },
  )
  }
  return logicalIds
@@ -140,14 +142,14 @@ internal class DatabaseImpl(
  resourceId: String,
  resourceType: ResourceType,
  versionId: String,
- lastUpdated: Instant
+ lastUpdated: Instant,
  ) {
  db.withTransaction {
  resourceDao.updateAndIndexRemoteVersionIdAndLastUpdate(
  resourceId,
  resourceType,
  versionId,
- lastUpdated
+ lastUpdated,
  )
  }
  }
@@ -174,12 +176,13 @@ internal class DatabaseImpl(
  null
  }
  val rowsDeleted = resourceDao.deleteResource(resourceId = id, resourceType = type)
- if (rowsDeleted > 0)
+ if (rowsDeleted > 0) {
  localChangeDao.addDelete(
  resourceId = id,
  resourceType = type,
- remoteVersionId = remoteVersionId
+ remoteVersionId = remoteVersionId,
  )
+ }
  }
  }
 
@@ -200,7 +203,7 @@ internal class DatabaseImpl(
  IndexedIdAndResource(
  it.matchingIndex,
  it.idOfBaseResourceOnWhichThisMatchedInc ?: it.idOfBaseResourceOnWhichThisMatchedRev!!,
- iParser.parseResource(it.serializedResource) as Resource
+ iParser.parseResource(it.serializedResource) as Resource,
  )
  }
  }
@@ -216,6 +219,10 @@ internal class DatabaseImpl(
  return db.withTransaction { localChangeDao.getAllLocalChanges().map { it.toLocalChange() } }
  }
 
+ override suspend fun getLocalChangesCount(): Int {
+ return db.withTransaction { localChangeDao.getLocalChangesCount() }
+ }
+
  override suspend fun deleteUpdates(token: LocalChangeToken) {
  db.withTransaction { localChangeDao.discardLocalChanges(token) }
  }
@@ -265,12 +272,12 @@ internal class DatabaseImpl(
  if (forcePurge) {
  resourceDao.deleteResource(resourceId = id, resourceType = type)
  localChangeDao.discardLocalChanges(
- token = LocalChangeToken(localChangeEntityList.map { it.id })
+ token = LocalChangeToken(localChangeEntityList.map { it.id }),
  )
  } else {
  // local change is available but FORCE_PURGE = false then throw exception
  throw IllegalStateException(
- "Resource with type $type and id $id has local changes, either sync with server or FORCE_PURGE required"
+ "Resource with type $type and id $id has local changes, either sync with server or FORCE_PURGE required",
  )
  }
  }
@@ -301,5 +308,5 @@ internal class DatabaseImpl(
 data class DatabaseConfig(
  val inMemory: Boolean,
  val enableEncryption: Boolean,
- val databaseErrorStrategy: DatabaseErrorStrategy
+ val databaseErrorStrategy: DatabaseErrorStrategy,
 )