Implement incremental LOI sync

app/src/main/java/org/groundplatform/android/data/local/room/dao/LocationOfInterestMutationDao.kt

@@ -41,16 +41,16 @@ interface LocationOfInterestMutationDao : BaseDao<LocationOfInterestMutationEnti
     vararg allowedStates: MutationEntitySyncStatus,
   ): List<LocationOfInterestMutationEntity>
 
-  /** Returns how many of the survey's LOIs hold a mutation of another type in one of the states. */
+  /** Returns how many of the survey's LOIs have a [type] mutation in one of [allowedStates]. */
   @Query(
     "SELECT COUNT(DISTINCT location_of_interest_id) FROM location_of_interest_mutation " +
       "WHERE survey_id = :surveyId " +
-      "AND type != :excludedType " +
+      "AND type = :type " +
       "AND state IN (:allowedStates)"
   )
   suspend fun countLocationOfInterestIds(
     surveyId: String,
-    excludedType: MutationEntityType,
+    type: MutationEntityType,
     vararg allowedStates: MutationEntitySyncStatus,
   ): Int
 }

app/src/main/java/org/groundplatform/android/data/local/room/stores/RoomLocationOfInterestStore.kt

@@ -115,6 +115,19 @@ class RoomLocationOfInterestStore @Inject internal constructor() : LocalLocation
     locationOfInterestDao.findById(locationOfInterestId)?.let { locationOfInterestDao.delete(it) }
   }
 
+  override suspend fun safeDeleteLocalLoi(locationOfInterestId: String) {
+    // One transaction, so a mutation saved right now can't be cascaded away by the delete.
+    localDatabase.withTransaction {
+      val pending =
+        locationOfInterestMutationDao.getMutations(
+          locationOfInterestId,
+          MutationEntitySyncStatus.PENDING,
+          MutationEntitySyncStatus.IN_PROGRESS,
+        )
+      if (pending.isEmpty()) deleteLocationOfInterest(locationOfInterestId)
+    }
+  }
+
   override fun getAllSurveyMutations(survey: Survey): Flow<List<LocationOfInterestMutation>> =
     locationOfInterestMutationDao.getAllMutationsFlow().map { mutations ->
       mutations.filter { it.surveyId == survey.id }.map { it.toModelObject() }
@@ -144,15 +157,15 @@ class RoomLocationOfInterestStore @Inject internal constructor() : LocalLocation
     locationOfInterestDao.upsertAll(entities)
   }
 
-  override suspend fun countPendingNonDeletedLois(surveyId: String): Int =
+  override suspend fun countPendingCreatedLois(surveyId: String): Int =
     locationOfInterestMutationDao.countLocationOfInterestIds(
       surveyId,
-      MutationEntityType.DELETE,
+      MutationEntityType.CREATE,
       MutationEntitySyncStatus.PENDING,
       MutationEntitySyncStatus.IN_PROGRESS,
     )
 
-  override suspend fun deleteNotIn(surveyId: String, ids: List<String>) {
+  override suspend fun deleteNotIn(surveyId: String, ids: Collection<String>) {
     val idsToKeep = ids.toSet()
     localDatabase.withTransaction {
       locationOfInterestDao

app/src/main/java/org/groundplatform/android/data/local/room/stores/RoomSurveySyncStateStore.kt

@@ -17,42 +17,35 @@ package org.groundplatform.android.data.local.room.stores
 
 import javax.inject.Inject
 import kotlin.time.Clock
+import org.groundplatform.android.data.local.room.converter.toLocalDataStoreObject
 import org.groundplatform.android.data.local.room.converter.toModelObject
 import org.groundplatform.android.data.local.room.dao.SurveySyncStateDao
 import org.groundplatform.android.data.local.room.dao.insertOrUpdate
-import org.groundplatform.android.data.local.room.entity.SurveySyncStateEntity
 import org.groundplatform.android.data.local.stores.LocalSurveySyncStateStore
-import org.groundplatform.android.data.remote.firebase.protobuf.toProto
 import org.groundplatform.domain.model.Survey
 import org.groundplatform.domain.model.SurveySyncState
 
 class RoomSurveySyncStateStore
 @Inject
 constructor(private val surveySyncStateDao: SurveySyncStateDao) : LocalSurveySyncStateStore {
-  override suspend fun get(surveyId: String): SurveySyncState? {
-    val entity = surveySyncStateDao.get(surveyId)
-    return entity?.toModelObject()
-  }
+  override suspend fun get(surveyId: String): SurveySyncState? =
+    surveySyncStateDao.get(surveyId)?.toModelObject()
 
-  override suspend fun recordIncrementalSync(
-    surveyId: String,
-    latestLoiServerTimestamp: Long,
-  ) {
+  override suspend fun recordIncrementalSync(surveyId: String, latestLoiServerTimestamp: Long) =
     surveySyncStateDao.updateLatestLoiServerTimestamp(surveyId, latestLoiServerTimestamp)
-  }
 
   override suspend fun recordFullSync(
     surveyId: String,
     latestLoiServerTimestamp: Long,
     dataVisibility: Survey.DataVisibility?,
-  ) {
+  ) =
     surveySyncStateDao.insertOrUpdate(
-      SurveySyncStateEntity(
-        surveyId = surveyId,
-        latestLoiServerTimestamp = latestLoiServerTimestamp,
-        lastFullSyncClientTimestamp = Clock.System.now().toEpochMilliseconds(),
-        syncedDataVisibility = dataVisibility?.toProto()?.ordinal,
-      )
+      SurveySyncState(
+          surveyId = surveyId,
+          latestLoiServerTimestamp = latestLoiServerTimestamp,
+          lastFullSyncClientTimestamp = Clock.System.now().toEpochMilliseconds(),
+          syncedDataVisibility = dataVisibility,
+        )
+        .toLocalDataStoreObject()
     )
-  }
 }

app/src/main/java/org/groundplatform/android/data/local/stores/LocalLocationOfInterestStore.kt

@@ -46,6 +46,9 @@ interface LocalLocationOfInterestStore :
   /** Deletes LOI from local database. */
   suspend fun deleteLocationOfInterest(locationOfInterestId: String)
 
+  /** Deletes LOI from local database, keeping it if it has changes still waiting to upload. */
+  suspend fun safeDeleteLocalLoi(locationOfInterestId: String)
+
   /**
    * Returns a [Flow] that emits a [List] of all [LocationOfInterestMutation]s stored in the local
    * db related to a given [Survey]. A new [List] is emitted on each change to the underlying saved
@@ -65,11 +68,8 @@ interface LocalLocationOfInterestStore :
   /** Inserts or updates all the given LOIs in a single transaction. */
   suspend fun insertOrUpdateAll(lois: List<LocationOfInterest>)
 
-  suspend fun deleteNotIn(surveyId: String, ids: List<String>)
+  suspend fun deleteNotIn(surveyId: String, ids: Collection<String>)
 
-  /**
-   * Returns the number of survey LOIs with a pending local change that has not yet been synced,
-   * excluding deletes.
-   */
-  suspend fun countPendingNonDeletedLois(surveyId: String): Int
+  /** Returns how many of the survey's LOIs were created locally and not uploaded yet. */
+  suspend fun countPendingCreatedLois(surveyId: String): Int
 }

app/src/main/java/org/groundplatform/android/data/remote/RemoteDataStore.kt

@@ -50,15 +50,22 @@ interface RemoteDataStore {
   suspend fun loadTermsOfService(): TermsOfService?
 
   /** Returns predefined LOIs in the specified survey. Main-safe. */
-  fun loadPredefinedLois(survey: Survey): Flow<List<LocationOfInterest>>
+  fun loadPredefinedLois(survey: Survey, fromTimestamp: Long?): Flow<List<LocationOfInterest>>
 
   /** Returns LOIs owned by the specified user in the specified survey. Main-safe. */
-  fun loadUserLois(survey: Survey, ownerUserId: String): Flow<List<LocationOfInterest>>
+  fun loadUserLois(
+    survey: Survey,
+    ownerUserId: String,
+    fromTimestamp: Long?,
+  ): Flow<List<LocationOfInterest>>
 
   /**
    * Returns LOIs that have been marked as shared for other participants of the specified survey.
    */
-  fun loadSharedLois(survey: Survey): Flow<List<LocationOfInterest>>
+  fun loadSharedLois(survey: Survey, fromTimestamp: Long?): Flow<List<LocationOfInterest>>
+
+  /** Returns how many LOIs a sync of the specified survey would fetch. Main-safe. */
+  suspend fun countLois(survey: Survey, ownerUserId: String): Long
 
   /**
    * Applies the provided mutations to the remote data store in a single batched transaction. If one

app/src/main/java/org/groundplatform/android/data/remote/firebase/FirebaseMessagingService.kt

@@ -20,11 +20,18 @@ import com.google.firebase.messaging.FirebaseMessagingService
 import com.google.firebase.messaging.RemoteMessage
 import dagger.hilt.android.AndroidEntryPoint
 import javax.inject.Inject
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.launch
+import org.groundplatform.android.data.local.stores.LocalLocationOfInterestStore
 import org.groundplatform.android.data.sync.SurveySyncService
+import org.groundplatform.android.di.coroutines.ApplicationScope
 import timber.log.Timber
 
 const val TOPIC_PREFIX = "/topics/"
 
+private const val LOI_ID_KEY = "loiId"
+private const val DELETED_KEY = "deleted"
+
 /**
  * Listens to messages from Firebase Cloud Messaging, and enqueuing re-sync of survey metadata when
  * receiving.
@@ -33,6 +40,8 @@ const val TOPIC_PREFIX = "/topics/"
 class FirebaseMessagingService : FirebaseMessagingService() {
 
   @Inject lateinit var surveySyncService: SurveySyncService
+  @Inject lateinit var localLoiStore: LocalLocationOfInterestStore
+  @Inject @ApplicationScope lateinit var externalScope: CoroutineScope
 
   /**
    * Processes new messages, enqueuing a worker to sync the survey with the id specified in the
@@ -45,6 +54,12 @@ class FirebaseMessagingService : FirebaseMessagingService() {
       return
     }
     Timber.v("Message received from topic ${remoteMessage.from}")
+
+    // Dropping it here spares the sync the full read it would take to notice the deletion.
+    remoteMessage.data[LOI_ID_KEY]
+      ?.takeIf { remoteMessage.data[DELETED_KEY].toBoolean() }
+      ?.let { externalScope.launch { localLoiStore.safeDeleteLocalLoi(it) } }
+
     surveySyncService.enqueueSync(surveyId)
   }
 

app/src/main/java/org/groundplatform/android/data/remote/firebase/FirestoreDataStore.kt

@@ -23,6 +23,8 @@ import javax.inject.Inject
 import javax.inject.Singleton
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.emitAll
@@ -35,6 +37,7 @@ import org.groundplatform.android.BuildConfig.USE_EMULATORS
 import org.groundplatform.android.data.remote.RemoteDataStore
 import org.groundplatform.android.data.remote.firebase.schema.GroundFirestore
 import org.groundplatform.android.data.remote.firebase.schema.LoiCollectionReference
+import org.groundplatform.android.data.remote.firebase.schema.LoiQueryScope
 import org.groundplatform.android.di.coroutines.IoDispatcher
 import org.groundplatform.domain.model.Survey
 import org.groundplatform.domain.model.SurveyListItem
@@ -84,13 +87,29 @@ internal constructor(
     )
   }
 
-  override fun loadPredefinedLois(survey: Survey) =
-    fetchLoiPages(survey) { fetchPredefined(survey) }
-
-  override fun loadUserLois(survey: Survey, ownerUserId: String) =
-    fetchLoiPages(survey) { fetchUserDefined(survey, ownerUserId) }
-
-  override fun loadSharedLois(survey: Survey) = fetchLoiPages(survey) { fetchSharedLois(survey) }
+  override fun loadPredefinedLois(survey: Survey, fromTimestamp: Long?) =
+    fetchLoiPages(survey) { fetch(survey, LoiQueryScope.Predefined, fromTimestamp) }
+
+  override fun loadUserLois(survey: Survey, ownerUserId: String, fromTimestamp: Long?) =
+    fetchLoiPages(survey) { fetch(survey, LoiQueryScope.UserDefined(ownerUserId), fromTimestamp) }
+
+  override fun loadSharedLois(survey: Survey, fromTimestamp: Long?) =
+    fetchLoiPages(survey) { fetch(survey, LoiQueryScope.Shared, fromTimestamp) }
+
+  override suspend fun countLois(survey: Survey, ownerUserId: String): Long =
+    withContext(ioDispatcher) {
+      val lois = db().surveys().survey(survey.id).lois()
+      val fieldData =
+        if (survey.dataVisibility == Survey.DataVisibility.ALL_SURVEY_PARTICIPANTS) {
+          LoiQueryScope.Shared
+        } else {
+          LoiQueryScope.UserDefined(ownerUserId)
+        }
+      // A round trip each, and neither needs the other's answer.
+      listOf(async { lois.count(LoiQueryScope.Predefined) }, async { lois.count(fieldData) })
+        .awaitAll()
+        .sum()
+    }
 
   /** Emits the pages of LOIs produced by [fetch] against the given survey's LOI collection. */
   private fun fetchLoiPages(