MS926 Upload count fixes

infra/event-sync/src/main/java/com/simprints/infra/eventsync/sync/up/tasks/EventUpSyncTask.kt

@@ -72,7 +72,6 @@ internal class EventUpSyncTask @Inject constructor(
         val project = projectWithConfig.project
         val config = projectWithConfig.configuration
         var lastOperation = operation.copy()
-        var count = 0
         var isUsefulUpload = false
 
         try {
@@ -95,8 +94,7 @@ internal class EventUpSyncTask @Inject constructor(
                     isUsefulUpload = it > 0
                     EventUpSyncRequestEvent.UpSyncContent(sessionCount = it)
                 },
-            ).collect {
-                count = it
+            ).collect { count ->
                 lastOperation = lastOperation.copy(
                     lastState = RUNNING,
                     lastSyncTime = timeHelper.now().ms,
@@ -112,8 +110,7 @@ internal class EventUpSyncTask @Inject constructor(
                     isUsefulUpload = it > 0
                     EventUpSyncRequestEvent.UpSyncContent(eventDownSyncCount = it)
                 },
-            ).collect {
-                count = it
+            ).collect { count ->
                 lastOperation = lastOperation.copy(
                     lastState = RUNNING,
                     lastSyncTime = timeHelper.now().ms,
@@ -131,8 +128,7 @@ internal class EventUpSyncTask @Inject constructor(
                         eventUpSyncCount = if (isUsefulUpload) it else 0,
                     )
                 },
-            ).collect {
-                count = it
+            ).collect { count ->
                 lastOperation = lastOperation.copy(
                     lastState = RUNNING,
                     lastSyncTime = timeHelper.now().ms,
@@ -144,7 +140,7 @@ internal class EventUpSyncTask @Inject constructor(
                 lastSyncTime = timeHelper.now().ms,
             )
 
-            emitProgress(lastOperation, count)
+            emitProgress(lastOperation, 0)
         } catch (t: Throwable) {
             if (t is RemoteDbNotSignedInException) {
                 throw t
@@ -156,7 +152,7 @@ internal class EventUpSyncTask @Inject constructor(
                 lastSyncTime = timeHelper.now().ms,
             )
 
-            emitProgress(lastOperation, count)
+            emitProgress(lastOperation, 0)
         }
     }
 
@@ -331,7 +327,7 @@ internal class EventUpSyncTask @Inject constructor(
         else -> emptyList()
     }
 
-    private suspend fun attemptInvalidEventUpload(
+    private fun attemptInvalidEventUpload(
         projectId: String,
         corruptedScopes: Set<EventScope>,
     ) = flow {

infra/event-sync/src/main/java/com/simprints/infra/eventsync/sync/up/workers/EventUpSyncUploaderWorker.kt

@@ -77,7 +77,7 @@ internal class EventUpSyncUploaderWorker @AssistedInject constructor(
             val workerId = this@EventUpSyncUploaderWorker.id.toString()
             var count = eventSyncCache.readProgress(workerId)
             val max = eventRepository
-                .observeEventCount(null)
+                .observeEventCountInClosedScopes()
                 .firstOrNull() ?: 0
 
             upSyncTask.upSync(upSyncScope.operation, getEventScope()).collect {

infra/event-sync/src/test/java/com/simprints/infra/eventsync/sync/up/workers/EventUpSyncUploaderWorkerTest.kt

@@ -73,7 +73,7 @@ internal class EventUpSyncUploaderWorkerTest {
 
         coEvery { eventSyncCache.readProgress(any()) } returns 0
         every { authStore.signedInProjectId } returns PROJECT_ID
-        coEvery { eventRepository.observeEventCount(any()) } returns flowOf(12)
+        coEvery { eventRepository.observeEventCountInClosedScopes() } returns flowOf(12)
     }
 
     @Test

infra/events/src/main/java/com/simprints/infra/events/EventRepository.kt

@@ -57,6 +57,8 @@ interface EventRepository {
 
     suspend fun observeEventCount(type: EventType?): Flow<Int>
 
+    suspend fun observeEventCountInClosedScopes(): Flow<Int>
+
     suspend fun addOrUpdateEvent(
         scope: EventScope,
         event: Event,

infra/events/src/main/java/com/simprints/infra/events/EventRepositoryImpl.kt

@@ -169,6 +169,8 @@ internal open class EventRepositoryImpl @Inject constructor(
         eventLocalDataSource.observeEventCount()
     }
 
+    override suspend fun observeEventCountInClosedScopes(): Flow<Int> = eventLocalDataSource.observeEventCountInClosedScopes()
+
     override suspend fun addOrUpdateEvent(
         scope: EventScope,
         event: Event,

infra/events/src/main/java/com/simprints/infra/events/event/local/EventLocalDataSource.kt

@@ -147,6 +147,10 @@ internal open class EventLocalDataSource @Inject constructor(
         eventDao.observeCountFromType(type = type)
     }
 
+    suspend fun observeEventCountInClosedScopes(): Flow<Int> = useRoomFlow(readingDispatcher) {
+        eventDao.observeCountInClosedScopes()
+    }
+
     suspend fun loadAllEvents(): Flow<Event> = useRoom(readingDispatcher) {
         eventDao.loadAll().map { it.fromDbToDomain() }.asFlow()
     }

infra/events/src/main/java/com/simprints/infra/events/event/local/EventRoomDao.kt

@@ -25,6 +25,15 @@ internal interface EventRoomDao {
     @Query("select count(*) from DbEvent where type = :type")
     fun observeCountFromType(type: EventType): Flow<Int>
 
+    @Query(
+        """
+        select count(*) from DbEvent 
+        left join DbEventScope on DbEvent.scopeId = DbEventScope.id 
+        where DbEventScope.end_unixMs is not null
+        """,
+    )
+    fun observeCountInClosedScopes(): Flow<Int>
+
     @Query("delete from DbEvent where scopeId = :scopeId")
     suspend fun deleteAllFromScope(scopeId: String)
 

infra/events/src/test/java/com/simprints/infra/events/EventRepositoryImplTest.kt

@@ -338,6 +338,15 @@ internal class EventRepositoryImplTest {
         coVerify(exactly = 1) { eventLocalDataSource.observeEventCount(any()) }
     }
 
+    @Test
+    fun `should delegate observeEventCountInClosedScopes calls`() = runTest {
+        coEvery { eventLocalDataSource.observeEventCountInClosedScopes() } returns flowOf(7)
+
+        assertThat(eventRepo.observeEventCountInClosedScopes().firstOrNull()).isEqualTo(7)
+
+        coVerify { eventLocalDataSource.observeEventCountInClosedScopes() }
+    }
+
     @Test
     fun `insert event into event scope should update event fields`() = runTest {
         val scope = createSessionScope(GUID1)

infra/events/src/test/java/com/simprints/infra/events/event/local/EventLocalDataSourceTest.kt

@@ -422,6 +422,13 @@ internal class EventLocalDataSourceTest {
         coVerify { eventDao.observeCount() }
     }
 
+    @Test
+    fun observeEventCountInClosedScopes() = runTest {
+        eventLocalDataSource.observeEventCountInClosedScopes().toList()
+
+        coVerify { eventDao.observeCountInClosedScopes() }
+    }
+
     @Test
     fun observeCountWithAProjectIdAndTypeQuery() = runTest {
         eventLocalDataSource