Adds Oura integration

src/main/kotlin/org/radarbase/push/integration/GarminPushIntegrationResourceEnhancer.kt

@@ -10,12 +10,15 @@ import org.radarbase.gateway.Config
 import org.radarbase.jersey.auth.AuthValidator
 import org.radarbase.jersey.enhancer.JerseyResourceEnhancer
 import org.radarbase.push.integration.common.auth.DelegatedAuthValidator.Companion.GARMIN_QUALIFIER
+import org.radarbase.push.integration.common.auth.DelegatedAuthValidator.Companion.OURA_QUALIFIER
 import org.radarbase.push.integration.common.user.User
 import org.radarbase.push.integration.garmin.auth.GarminAuthValidator
 import org.radarbase.push.integration.garmin.factory.GarminAuthMetadataFactory
 import org.radarbase.push.integration.garmin.factory.GarminUserTreeMapFactory
 import org.radarbase.push.integration.garmin.service.BackfillService
+import org.radarbase.push.integration.oura.service.OuraBackfillService
 import org.radarbase.push.integration.garmin.service.GarminHealthApiService
+import org.radarbase.push.integration.oura.user.OuraUserRepository
 import org.radarbase.push.integration.garmin.user.GarminUserRepository
 
 
@@ -31,18 +34,23 @@ class GarminPushIntegrationResourceEnhancer(private val config: Config) :
 
     override val classes: Array<Class<*>>
         get() = if (config.pushIntegration.garmin.backfill.enabled) {
-            arrayOf(BackfillService::class.java)
+            arrayOf(OuraBackfillService::class.java)
         } else {
             emptyArray()
         }
 
     override fun AbstractBinder.enhance() {
 
         bind(config.pushIntegration.garmin.userRepository)
-            .to(GarminUserRepository::class.java)
-            .named(GARMIN_QUALIFIER)
+        .to(GarminUserRepository::class.java)
+        .named(GARMIN_QUALIFIER)
             .`in`(Singleton::class.java)
 
+        bind(config.pushIntegration.garmin.userRepository)
+        .to(OuraUserRepository::class.java)
+        .named(OURA_QUALIFIER)
+        .`in`(Singleton::class.java)
+
         bind(GarminHealthApiService::class.java)
             .to(GarminHealthApiService::class.java)
             .`in`(Singleton::class.java)

src/main/kotlin/org/radarbase/push/integration/common/auth/DelegatedAuthValidator.kt

@@ -27,6 +27,7 @@ class DelegatedAuthValidator(
 
     companion object {
         const val GARMIN_QUALIFIER = "garmin"
+        const val OURA_QUALIFIER = "oura"
     }
 
     override fun verify(token: String, request: ContainerRequestContext): Auth? =

src/main/kotlin/org/radarbase/push/integration/oura/auth/OuraAuthValidator.kt

@@ -0,0 +1,128 @@
+package org.radarbase.push.integration.oura.auth
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import jakarta.inject.Named
+import jakarta.ws.rs.container.ContainerRequestContext
+import jakarta.ws.rs.core.Context
+import org.radarbase.jersey.auth.Auth
+import org.radarbase.jersey.auth.AuthValidator
+import org.radarbase.jersey.auth.disabled.DisabledAuth
+import org.radarbase.jersey.exception.HttpUnauthorizedException
+import org.radarbase.push.integration.common.auth.DelegatedAuthValidator.Companion.GARMIN_QUALIFIER
+import org.radarbase.oura.user.User
+import org.radarbase.push.integration.garmin.user.GarminUserRepository
+import org.radarbase.push.integration.oura.user.OuraServiceUserRepository
+import org.radarbase.push.integration.oura.user.OuraUserRepository
+import org.slf4j.LoggerFactory
+import java.time.Instant
+
+
+class OuraAuthValidator(
+    @Context private val objectMapper: ObjectMapper,
+    @Named(GARMIN_QUALIFIER) private val userRepository: OuraServiceUserRepository
+) :
+    AuthValidator {
+
+    private var nextRetry: Instant = Instant.MIN
+
+    override fun verify(token: String, request: ContainerRequestContext): Auth {
+        return if (token.isBlank()) {
+            throw HttpUnauthorizedException("invalid_token", "The token was empty")
+        } else {
+            var isAnyUnauthorised = false
+            // Enrich the request by adding the User
+            // the data format in Garmin's post is { <data-type> : [ {<data-1>}, {<data-2>} ] }
+            val tree = request.getProperty("tree") as JsonNode
+
+            val userTreeMap: Map<User, JsonNode> =
+                // group by user ID since request can contain data from multiple users
+                tree[tree.fieldNames().next()]
+                    .groupBy { node ->
+                        node[USER_ID_KEY].asText()
+                    }
+                    .filter { (userId, userData) ->
+                        val accessToken = userData[0][USER_ACCESS_TOKEN_KEY].asText()
+                        if (checkIsAuthorised(userId, accessToken)) true else {
+                            isAnyUnauthorised = true
+                            userRepository.deregisterUser(userId, accessToken)
+                            false
+                        }
+                    }
+                    .entries
+                    .associate { (userId, userData) ->
+                        userRepository.findByExternalId(userId) to
+                            // Map the List<JsonNode> back to <data-type>: [ {<data-1>}, {<data-2>} ]
+                            // so it can be processed in the services without much refactoring
+                            objectMapper.createObjectNode()
+                                .set(tree.fieldNames().next(), objectMapper.valueToTree(userData))
+                    }
+
+            request.setProperty("user_tree_map", userTreeMap)
+            request.setProperty(
+                "auth_metadata",
+                mapOf("isAnyUnauthorised" to isAnyUnauthorised.toString())
+            )
+            request.removeProperty("tree")
+
+            // Disable auth since we don't have proper auth support
+            DisabledAuth("res_gateway")
+        }
+    }
+
+    override fun getToken(request: ContainerRequestContext): String? {
+        return if (request.hasEntity()) {
+            // We put the json tree in the request because the entity stream will be closed here
+            val tree = objectMapper.readTree(request.entityStream)
+            request.setProperty("tree", tree)
+            val userAccessToken = tree[tree.fieldNames().next()][0][USER_ACCESS_TOKEN_KEY]
+                ?: throw HttpUnauthorizedException("invalid_token", "No user access token provided")
+            userAccessToken.asText().also {
+                request.setProperty(USER_ACCESS_TOKEN_KEY, it)
+            }
+        } else {
+            null
+        }
+    }
+
+    private fun checkIsAuthorised(userId: String, accessToken: String, retry: Boolean = true):
+        Boolean {
+        val user = try {
+            userRepository.findByExternalId(userId)
+        } catch (exc: NoSuchElementException) {
+            return if (retry && Instant.now() > nextRetry) {
+                userRepository.applyPendingUpdates()
+                nextRetry = Instant.now().plusSeconds(REFRESH_TIMEOUT_S)
+                checkIsAuthorised(userId, accessToken, retry = false)
+            } else {
+                logger.warn(
+                    "no_user: The user {} could not be found in the " +
+                        "user repository.", userId
+                )
+                false
+            }
+        }
+        if (!user.isAuthorized) {
+            logger.warn(
+                "invalid_user: The user {} does not seem to be authorized.", userId
+            )
+            return false
+        }
+        if (userRepository.getAccessToken(user) != accessToken) {
+            logger.warn(
+                "invalid_token: The token for user {} does not" +
+                    " match with the records on the system.", userId
+            )
+            return false
+        }
+        return true
+    }
+
+    companion object {
+        const val USER_ID_KEY = "userId"
+        const val USER_ACCESS_TOKEN_KEY = "userAccessToken"
+        const val REFRESH_TIMEOUT_S = 5L
+
+        private val logger = LoggerFactory.getLogger(OuraAuthValidator::class.java)
+    }
+}

src/main/kotlin/org/radarbase/push/integration/oura/backfill/OuraReqGenerator.kt

@@ -0,0 +1,62 @@
+package org.radarbase.push.integration.oura.backfill
+
+import okhttp3.Response
+import org.radarbase.gateway.Config
+import org.radarbase.oura.route.Route
+import org.radarbase.oura.user.User
+import org.radarbase.push.integration.garmin.backfill.route.*
+import org.radarbase.push.integration.oura.offset.OuraRedisOffsetManager
+import org.radarbase.push.integration.garmin.user.GarminUserRepository
+import org.radarbase.push.integration.oura.user.OuraUserRepository
+import org.radarbase.push.integration.garmin.util.RedisHolder
+import org.radarbase.push.integration.garmin.util.offset.*
+import org.slf4j.LoggerFactory
+import redis.clients.jedis.JedisPool
+import java.nio.file.Path
+import java.time.Duration
+import java.time.Instant
+import org.radarbase.oura.request.OuraRequestGenerator
+import org.radarbase.oura.request.RestRequest
+import org.radarbase.push.integration.garmin.backfill.TooManyRequestsException
+
+class OuraReqGenerator(
+    val config: Config,
+    private val userRepository: OuraUserRepository,
+    private val redisHolder: RedisHolder =
+        RedisHolder(JedisPool(config.pushIntegration.garmin.backfill.redis.uri)),
+    private val offsetPersistenceFactory: OffsetPersistenceFactory =
+        OffsetRedisPersistence(redisHolder),
+    private val defaultQueryRange: Duration = Duration.ofDays(15),
+) {
+
+    private val ouraOffsetManager = OuraRedisOffsetManager(redisUri = config.pushIntegration.garmin.backfill.redis.uri, redisHolder, offsetPersistenceFactory)
+
+    private var ouraRequestGenerator: OuraRequestGenerator = OuraRequestGenerator(userRepository = userRepository, ouraOffsetManager = ouraOffsetManager);
+
+    private val userNextRequest: MutableMap<String, Instant> = mutableMapOf()
+
+    private var nextRequestTime: Instant = Instant.MIN
+
+    private val shouldBackoff: Boolean
+        get() = Instant.now() < nextRequestTime
+
+    fun requests(user: User, max: Int): Sequence<RestRequest> {
+        val requests = ouraRequestGenerator.requests(user, max)
+        return requests
+    }
+
+    fun requestSuccessful(request: RestRequest, response: Response) {
+        val records = ouraRequestGenerator.requestSuccessful(request, response)
+        logger.info(records.toString())
+    }
+
+    fun requestFailed(request: RestRequest, response: Response) {
+        ouraRequestGenerator.requestFailed(request, response)
+    }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(OuraReqGenerator::class.java)
+        private val BACK_OFF_TIME = Duration.ofMinutes(1L)
+        private val USER_BACK_OFF_TIME = Duration.ofDays(1L)
+    }
+}

src/main/kotlin/org/radarbase/push/integration/oura/offset/OuraRedisOffsetManager.kt

@@ -0,0 +1,47 @@
+package org.radarbase.push.integration.oura.offset
+
+import okhttp3.Response
+import org.radarbase.oura.route.Route
+import org.radarbase.oura.user.User
+import org.slf4j.LoggerFactory
+import redis.clients.jedis.JedisPool
+import java.time.Instant
+import org.radarbase.oura.request.OuraOffsetManager
+import org.radarbase.oura.offset.Offset
+import org.radarbase.push.integration.garmin.util.RedisHolder
+import org.radarbase.push.integration.garmin.util.offset.*
+import java.nio.file.Path
+import java.net.URI
+
+class OuraRedisOffsetManager(
+    val redisUri: URI,
+    private val redisHolder: RedisHolder =
+        RedisHolder(JedisPool(redisUri)),
+    private val offsetPersistenceFactory: OffsetPersistenceFactory =
+        OffsetRedisPersistence(redisHolder),
+): OuraOffsetManager {
+
+    override fun getOffset(route: Route, user: User): Offset? {
+        logger.info("Getting offset..")
+        val offsets = offsetPersistenceFactory.read(user.versionedId)
+        if (offsets == null) return null
+        return Offset(user.userId, route.toString(), offsets.offsetsMap.getOrDefault(
+                            UserRoute(user.versionedId, route.toString()), user.startDate
+                        ).coerceAtLeast(user.startDate))
+    }
+
+    override fun updateOffsets(route: Route, user: User, offset: Instant) {
+        logger.info("Writing to offsets..")
+        offsetPersistenceFactory.add(
+            Path.of(user.versionedId), UserRouteOffset(
+                user.versionedId,
+                route.toString(),
+                offset
+            )
+        )
+    }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(OuraRedisOffsetManager::class.java)
+    }
+}