feat: updating

stellio-hub · Oct 16, 2024 · 8387130 · 8387130
1 parent e8ccf0c
commit 8387130
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 3 deletions.
diff --git a/search-service/src/main/kotlin/com/egm/stellio/search/csr/model/ContextSourceRegistration.kt b/search-service/src/main/kotlin/com/egm/stellio/search/csr/model/ContextSourceRegistration.kt
@@ -13,6 +13,7 @@ import com.egm.stellio.shared.util.JsonLdUtils.expandJsonLdTerm
 import com.egm.stellio.shared.util.JsonUtils.deserializeAs
 import com.egm.stellio.shared.util.JsonUtils.serializeObject
 import com.fasterxml.jackson.annotation.JsonFormat
+import com.fasterxml.jackson.annotation.JsonProperty
 import com.fasterxml.jackson.module.kotlin.convertValue
 import org.springframework.http.MediaType
 import java.net.URI
@@ -30,7 +31,13 @@ data class ContextSourceRegistration(
     val createdAt: ZonedDateTime = ngsiLdDateTime(),
     val modifiedAt: ZonedDateTime? = null,
     val observationInterval: TimeInterval? = null,
-    val managementInterval: TimeInterval? = null
+    val managementInterval: TimeInterval? = null,
+
+    var status: StatusType? = null,
+    val timesSent: Int = 0,
+    val timesFailed: Int = 0,
+    val lastFailure: ZonedDateTime? = null,
+    val lastSuccess: ZonedDateTime? = null,
 ) {
 
     data class TimeInterval(
@@ -156,6 +163,14 @@ data class ContextSourceRegistration(
         fun alreadyExistsMessage(id: URI) = "A CSourceRegistration with id $id already exists"
         fun unauthorizedMessage(id: URI) = "User is not authorized to access CSourceRegistration $id"
     }
+
+    enum class StatusType(val status: String) {
+        @JsonProperty("ok")
+        OK("ok"),
+
+        @JsonProperty("failed")
+        FAILED("failed")
+    }
 }
 
 fun List<ContextSourceRegistration>.serialize(

diff --git a/...ce/src/main/kotlin/com/egm/stellio/search/csr/service/ContextSourceRegistrationService.kt b/...ce/src/main/kotlin/com/egm/stellio/search/csr/service/ContextSourceRegistrationService.kt
@@ -16,14 +16,18 @@ import com.egm.stellio.shared.util.Sub
 import com.egm.stellio.shared.util.mapper
 import com.egm.stellio.shared.util.toStringValue
 import io.r2dbc.postgresql.codec.Json
+import kotlinx.coroutines.reactive.awaitFirst
 import org.springframework.data.r2dbc.core.R2dbcEntityTemplate
 import org.springframework.data.relational.core.query.Criteria.where
 import org.springframework.data.relational.core.query.Query.query
+import org.springframework.data.relational.core.query.Update
 import org.springframework.r2dbc.core.DatabaseClient
 import org.springframework.r2dbc.core.bind
 import org.springframework.stereotype.Component
 import org.springframework.transaction.annotation.Transactional
 import java.net.URI
+import java.time.Instant
+import java.time.ZoneOffset
 
 @Component
 class ContextSourceRegistrationService(
@@ -234,4 +238,24 @@ class ContextSourceRegistrationService(
             },
         )
     }
+
+    suspend fun updateContextSourceStatus(
+        csr: ContextSourceRegistration,
+        success: Boolean
+    ): Long {
+        val updateStatement = if (success)
+            Update.update("status", ContextSourceRegistration.StatusType.OK.name)
+                .set("times_sent", csr.timesSent + 1)
+                .set("last_success", Instant.now().atZone(ZoneOffset.UTC))
+        else Update.update("status", ContextSourceRegistration.StatusType.FAILED.name)
+            .set("times_sent", csr.timesSent + 1)
+            .set("times_failed", csr.timesFailed + 1)
+            .set("last_failure", Instant.now().atZone(ZoneOffset.UTC))
+
+        return r2dbcEntityTemplate.update(
+            query(where("id").`is`(csr.id)),
+            updateStatement,
+            ContextSourceRegistration::class.java
+        ).awaitFirst()
+    }
 }
diff --git a/search-service/src/main/kotlin/com/egm/stellio/search/entity/web/EntityHandler.kt b/search-service/src/main/kotlin/com/egm/stellio/search/entity/web/EntityHandler.kt
@@ -219,12 +219,14 @@ class EntityHandler(
 
         // we can add parMap(concurrency = X) if this trigger too much http connexion at the same time
         val (entitiesWithMode, warnings) = matchingCSR.parMap { csr ->
-            ContextSourceUtils.getDistributedInformation(
+            val response = ContextSourceUtils.getDistributedInformation(
                 httpHeaders,
                 csr,
                 "/ngsi-ld/v1/entities/$entityId",
                 params
-            ) to csr.mode
+            )
+            contextSourceRegistrationService.updateContextSourceStatus(csr, response.isRight())
+            response to csr.mode
         }.partition { it.first.isRight() }
             .let { (responses, warnings) ->
                 responses.map { (response, mode) -> response.getOrNull()!! to mode } to