[VIT-6439] Handle end_date in daily active state (#116)

* [VIT-6439] Handle end_date in daily active state

.gitignore

@@ -10,6 +10,8 @@
 /.idea/androidTestResultsUserPreferences.xml
 /.idea/deploymentTargetDropDown.xml
 /.idea/inspectionProfiles/Project_Default.xml
+.idea/vcs.xml
+.idea/misc.xml
 .DS_Store
 /build
 /captures

VitalClient/src/main/java/io/tryvital/client/VitalClient.kt

@@ -175,7 +175,7 @@ class VitalClient internal constructor(context: Context) {
     }
 
     companion object {
-        const val sdkVersion = "2.0.3"
+        const val sdkVersion = "2.0.4"
 
         private var sharedInstance: VitalClient? = null
 

VitalHealthConnect/src/main/java/io/tryvital/vitalhealthconnect/workers/ResourceSyncWorker.kt

@@ -77,8 +77,10 @@ internal sealed class ResourceSyncState {
     }
 
     @JsonClass(generateAdapter = true)
-    data class Incremental(val changesToken: String, val lastSync: Date) : ResourceSyncState() {
-        override fun toString(): String = "incremental($changesToken at ${lastSync.toInstant()})"
+    data class Incremental(val changesToken: String, val lastSync: Date, val end: Date? = null) :
+        ResourceSyncState() {
+        override fun toString(): String =
+            "incremental($changesToken at ${lastSync.toInstant()}; end = ${end})"
     }
 
     companion object {
@@ -165,13 +167,17 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
                 vitalLogger.logI("${input.resource}: skipped because backend status is ${backendState.status}")
                 return Result.success()
             }
+
             UserSDKSyncStatus.Active -> when (state) {
                 // Prefer backend provided pull range if present.
                 is ResourceSyncState.Historical -> state.copy(
                     start = backendState.requestStartDate ?: state.start,
                     end = backendState.requestEndDate ?: state.end,
                 )
-                is ResourceSyncState.Incremental -> state
+
+                is ResourceSyncState.Incremental -> state.copy(
+                    end = backendState.requestEndDate
+                )
             }
         }
 
@@ -189,11 +195,17 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
     private fun initialState(timeZone: TimeZone): ResourceSyncState {
         val now = ZonedDateTime.now(timeZone.toZoneId())
         val start = now.minus(30, ChronoUnit.DAYS)
-        return ResourceSyncState.Historical(start = start.toInstant().toDate(), end = now.toInstant().toDate())
+        return ResourceSyncState.Historical(
+            start = start.toInstant().toDate(),
+            end = now.toInstant().toDate()
+        )
     }
 
     @OptIn(VitalPrivateApi::class)
-    private suspend fun fetchBackendSyncState(state: ResourceSyncState, timeZone: TimeZone): UserSDKSyncStateResponse {
+    private suspend fun fetchBackendSyncState(
+        state: ResourceSyncState,
+        timeZone: TimeZone
+    ): UserSDKSyncStateResponse {
         val backendState = vitalClient.vitalPrivateService.healthConnectSdkSyncState(
             vitalClient.checkUserId(),
             when (state) {
@@ -203,6 +215,7 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
                     requestStartDate = state.start,
                     requestEndDate = state.end,
                 )
+
                 is ResourceSyncState.Incremental -> UserSDKSyncStateBody(
                     stage = DataStage.Daily,
                     tzinfo = timeZone.id,
@@ -215,7 +228,10 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
         return backendState
     }
 
-    private suspend fun historicalBackfill(state: ResourceSyncState.Historical, timeZone: TimeZone) {
+    private suspend fun historicalBackfill(
+        state: ResourceSyncState.Historical,
+        timeZone: TimeZone
+    ) {
         genericBackfill(
             stage = DataStage.Historical,
             start = state.start.toInstant(),
@@ -224,7 +240,10 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
         )
     }
 
-    private suspend fun incrementalBackfill(state: ResourceSyncState.Incremental, timeZone: TimeZone) {
+    private suspend fun incrementalBackfill(
+        state: ResourceSyncState.Incremental,
+        timeZone: TimeZone
+    ) {
         val userId = vitalClient.checkUserId()
         val client = healthConnectClientProvider.getHealthConnectClient(applicationContext)
 
@@ -240,7 +259,7 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
             return genericBackfill(
                 stage = DataStage.Daily,
                 start = state.lastSync.toInstant(),
-                end = Instant.now(),
+                end = minOf(Instant.now(), state.end?.toInstant() ?: Instant.now()),
                 timeZone = timeZone,
             )
         }
@@ -257,7 +276,7 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
                 return genericBackfill(
                     stage = DataStage.Daily,
                     start = state.lastSync.toInstant(),
-                    end = Instant.now(),
+                    end = minOf(Instant.now(), state.end?.toInstant() ?: Instant.now()),
                     timeZone = timeZone,
                 )
             }
@@ -272,6 +291,7 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
                     currentDevice = Build.MODEL,
                     reader = recordReader,
                     processor = recordProcessor,
+                    end = state.end?.toInstant()
                 )
 
                 // Skip empty POST requests
@@ -297,7 +317,12 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
         } while (changes.hasMore)
     }
 
-    private suspend fun genericBackfill(stage: DataStage, start: Instant, end: Instant, timeZone: TimeZone) {
+    private suspend fun genericBackfill(
+        stage: DataStage,
+        start: Instant,
+        end: Instant,
+        timeZone: TimeZone
+    ) {
         val userId = vitalClient.checkUserId()
         val client = healthConnectClientProvider.getHealthConnectClient(applicationContext)
 
@@ -401,16 +426,20 @@ internal class ResourceSyncWorker(appContext: Context, workerParams: WorkerParam
     }
 }
 
-internal inline fun <reified T: Any> SharedPreferences.getJson(key: String): T?
-    = getJson(key, default = null)
+internal inline fun <reified T : Any> SharedPreferences.getJson(key: String): T? =
+    getJson(key, default = null)
 
 internal inline fun <reified T> SharedPreferences.getJson(key: String, default: T): T {
     val jsonString = getString(key, null) ?: return default
     val adapter = moshi.adapter(T::class.java)
-    return adapter.fromJson(jsonString) ?: throw IllegalStateException("Failed to decode JSON string")
+    return adapter.fromJson(jsonString)
+        ?: throw IllegalStateException("Failed to decode JSON string")
 }
 
-internal inline fun <reified T: Any> SharedPreferences.Editor.putJson(key: String, value: T?): SharedPreferences.Editor {
+internal inline fun <reified T : Any> SharedPreferences.Editor.putJson(
+    key: String,
+    value: T?
+): SharedPreferences.Editor {
     val adapter = moshi.adapter(T::class.java)
     return putString(key, value?.let(adapter::toJson))
 }
@@ -419,5 +448,5 @@ internal val VitalResource.syncStateKey get() = UnSecurePrefKeys.syncStateKey(th
 internal val VitalResource.monitoringTypesKey get() = UnSecurePrefKeys.monitoringTypesKey(this)
 
 // All Record types are public JVM types, so they must have a simple name.
-private fun Set<KClass<out Record>>.toSimpleNameSet(): Set<String>
-    = mapTo(mutableSetOf()) { it.simpleName!! }
+private fun Set<KClass<out Record>>.toSimpleNameSet(): Set<String> =
+    mapTo(mutableSetOf()) { it.simpleName!! }

VitalHealthConnect/src/main/java/io/tryvital/vitalhealthconnect/workers/processChangesResponse.kt

@@ -1,15 +1,31 @@
 package io.tryvital.vitalhealthconnect.workers
 
 import androidx.health.connect.client.changes.UpsertionChange
+import androidx.health.connect.client.records.ActiveCaloriesBurnedRecord
+import androidx.health.connect.client.records.BasalMetabolicRateRecord
+import androidx.health.connect.client.records.BloodGlucoseRecord
+import androidx.health.connect.client.records.BloodPressureRecord
+import androidx.health.connect.client.records.BodyFatRecord
+import androidx.health.connect.client.records.DistanceRecord
+import androidx.health.connect.client.records.ExerciseSessionRecord
+import androidx.health.connect.client.records.FloorsClimbedRecord
+import androidx.health.connect.client.records.HeartRateRecord
+import androidx.health.connect.client.records.HeartRateVariabilityRmssdRecord
+import androidx.health.connect.client.records.HeightRecord
+import androidx.health.connect.client.records.HydrationRecord
 import androidx.health.connect.client.records.Record
 import androidx.health.connect.client.records.SleepSessionRecord
+import androidx.health.connect.client.records.StepsRecord
+import androidx.health.connect.client.records.Vo2MaxRecord
+import androidx.health.connect.client.records.WeightRecord
 import androidx.health.connect.client.response.ChangesResponse
 import io.tryvital.vitalhealthconnect.model.VitalResource
 import io.tryvital.vitalhealthconnect.model.processedresource.ProcessedResourceData
 import io.tryvital.vitalhealthconnect.model.processedresource.TimeSeriesData
 import io.tryvital.vitalhealthconnect.model.remapped
 import io.tryvital.vitalhealthconnect.records.RecordProcessor
 import io.tryvital.vitalhealthconnect.records.RecordReader
+import java.time.Instant
 import java.util.*
 import kotlin.reflect.KClass
 
@@ -20,13 +36,16 @@ internal suspend fun processChangesResponse(
     currentDevice: String,
     reader: RecordReader,
     processor: RecordProcessor,
+    end: Instant? = null,
 ): ProcessedResourceData {
     val records = responses.changes
         .filterIsInstance<UpsertionChange>()
         .groupBy(keySelector = { it.record::class }, valueTransform = { it.record })
 
+    val endAdjusted = end ?: Instant.MAX;
 
-    suspend fun <Record, T: TimeSeriesData> readTimeseries(
+
+    suspend fun <Record, T : TimeSeriesData> readTimeseries(
         records: List<Record>,
         process: suspend (String, List<Record>) -> T
     ): ProcessedResourceData = process(currentDevice, records)
@@ -39,20 +58,24 @@ internal suspend fun processChangesResponse(
         VitalResource.Activity -> processor.processActivitiesFromRecords(
             timeZone = timeZone,
             currentDevice = currentDevice,
-            activeEnergyBurned = records.get(),
-            basalMetabolicRate = records.get(),
-            floorsClimbed = records.get(),
-            distance = records.get(),
-            steps = records.get(),
-            vo2Max = records.get(),
+            activeEnergyBurned = records.get<ActiveCaloriesBurnedRecord>()
+                .filter { it.endTime <= endAdjusted },
+            basalMetabolicRate = records.get<BasalMetabolicRateRecord>()
+                .filter { it.time <= endAdjusted },
+            floorsClimbed = records.get<FloorsClimbedRecord>().filter { it.endTime <= endAdjusted },
+            distance = records.get<DistanceRecord>().filter { it.endTime <= endAdjusted },
+            steps = records.get<StepsRecord>().filter { it.endTime <= endAdjusted },
+            vo2Max = records.get<Vo2MaxRecord>().filter { it.time <= endAdjusted },
         ).let(ProcessedResourceData::Summary)
 
         VitalResource.Workout -> processor.processWorkoutsFromRecords(
             fallbackDeviceModel = currentDevice,
-            exerciseRecords = records.get()
+            exerciseRecords = records.get<ExerciseSessionRecord>()
+                .filter { it.endTime <= endAdjusted }
         ).let(ProcessedResourceData::Summary)
 
-        VitalResource.Sleep -> records.get<SleepSessionRecord>().let { sessions ->
+        VitalResource.Sleep -> records.get<SleepSessionRecord>()
+            .filter { it.endTime <= endAdjusted }.let { sessions ->
             processor.processSleepFromRecords(
                 fallbackDeviceModel = currentDevice,
                 sleepSessionRecords = sessions,
@@ -61,28 +84,47 @@ internal suspend fun processChangesResponse(
 
         VitalResource.Body -> processor.processBodyFromRecords(
             fallbackDeviceModel = currentDevice,
-            weightRecords = records.get(),
-            bodyFatRecords = records.get(),
+            weightRecords = records.get<WeightRecord>().filter { it.time <= endAdjusted },
+            bodyFatRecords = records.get<BodyFatRecord>().filter { it.time <= endAdjusted },
         ).let(ProcessedResourceData::Summary)
 
         VitalResource.Profile -> processor.processProfileFromRecords(
-            heightRecords = records.get()
+            heightRecords = records.get<HeightRecord>().filter { it.time <= endAdjusted }
         ).let(ProcessedResourceData::Summary)
 
         VitalResource.HeartRate ->
-            readTimeseries(records.get(), processor::processHeartRateFromRecords)
+            readTimeseries(
+                records.get<HeartRateRecord>().filter { it.endTime <= endAdjusted },
+                processor::processHeartRateFromRecords
+            )
+
         VitalResource.HeartRateVariability ->
-            readTimeseries(records.get(), processor::processHeartRateVariabilityRmssFromRecords)
+            readTimeseries(
+                records.get<HeartRateVariabilityRmssdRecord>().filter { it.time <= endAdjusted },
+                processor::processHeartRateVariabilityRmssFromRecords
+            )
+
         VitalResource.Glucose ->
-            readTimeseries(records.get(), processor::processGlucoseFromRecords)
+            readTimeseries(
+                records.get<BloodGlucoseRecord>().filter { it.time <= endAdjusted },
+                processor::processGlucoseFromRecords
+            )
+
         VitalResource.BloodPressure ->
-            readTimeseries(records.get(), processor::processBloodPressureFromRecords)
+            readTimeseries(
+                records.get<BloodPressureRecord>().filter { it.time <= endAdjusted },
+                processor::processBloodPressureFromRecords
+            )
+
         VitalResource.Water ->
-            readTimeseries(records.get(), processor::processWaterFromRecords)
+            readTimeseries(
+                records.get<HydrationRecord>().filter { it.endTime <= endAdjusted },
+                processor::processWaterFromRecords
+            )
     }
 }
 
-inline fun <reified T: Record> Map<KClass<out Record>, List<Record>>.get(): List<T> {
+inline fun <reified T : Record> Map<KClass<out Record>, List<Record>>.get(): List<T> {
     @Suppress("UNCHECKED_CAST")
     return (this[T::class] ?: emptyList()) as List<T>
 }