AbstractExposedKeyValuesRepo#set fix

CHANGELOG.md

@@ -4,9 +4,13 @@
 
 * `Versions`:
   * `SQLite`: `3.49.0.0` -> `3.49.1.0`
+* `Common`:
+  * Add `retryOnFailure` utility for simple retries code writing
 * `Repos`:
   * `Cache`:
     * Fix of `FullKeyValueCacheRepo` fields usage
+  * `Exposed`:
+    * `AbstractExposedKeyValuesRepo` will produce `onValueRemoved` event on `set` if some data has been removed
 
 ## 0.24.6
 

common/src/commonMain/kotlin/dev/inmo/micro_utils/common/Retry.kt

@@ -0,0 +1,26 @@
+package dev.inmo.micro_utils.common
+
+/**
+ * Will try to execute [action] and, if any exception will happen, execution will be retried.
+ * This process will happen at most [count] times. There is no any limits on [count] value, but [action] will run at
+ * least once and [retryOnFailure] will return its result if it is successful
+ */
+inline fun <T> retryOnFailure(count: Int, action: () -> T): T {
+    var triesCount = 0
+    while (true) {
+        val result = runCatching {
+            action()
+        }.onFailure {
+            triesCount++
+
+            if (triesCount >= count) {
+                throw it
+            } else {
+                null
+            }
+        }
+
+        if (result.isSuccess) return result.getOrThrow()
+    }
+    error("Unreachable code: retry must throw latest exception if error happen or success value if not")
+}

repos/exposed/src/jvmMain/kotlin/dev/inmo/micro_utils/repos/exposed/onetomany/AbstractExposedKeyValuesRepo.kt

@@ -4,7 +4,6 @@ import dev.inmo.micro_utils.repos.KeyValuesRepo
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.*
 import org.jetbrains.exposed.sql.*
-import org.jetbrains.exposed.sql.statements.InsertStatement
 import org.jetbrains.exposed.sql.statements.UpdateBuilder
 import org.jetbrains.exposed.sql.transactions.transaction
 
@@ -57,18 +56,48 @@ abstract class AbstractExposedKeyValuesRepo<Key, Value>(
             }
         }
 
-        transaction(database) {
+        val (oldObjects, insertedResults) = transaction(database) {
+            val oldObjects = selectAll().where { selectByIds(toSet.keys.toList()) }.map { it.asKey to it.asObject }
+
             deleteWhere {
                 selectByIds(it, toSet.keys.toList())
             }
-            batchInsert(
+            val inserted = batchInsert(
                 prepreparedData,
             ) { (k, v) ->
                 insert(k, v, this)
             }.map {
                 it.asKey to it.asObject
             }
-        }.forEach { _onNewValue.emit(it) }
+            oldObjects to inserted
+        }.let {
+            val mappedFirst = it
+                .first
+                .asSequence()
+                .groupBy { it.first }
+                .mapValues { it.value.map { it.second }.toSet() }
+            val mappedSecond = it
+                .second
+                .asSequence()
+                .groupBy { it.first }
+                .mapValues { it.value.map { it.second }.toSet() }
+            mappedFirst to mappedSecond
+        }
+        val deletedResults = oldObjects.mapNotNull { (k, vs) ->
+            k to vs.filter { v ->
+                insertedResults[k] ?.contains(v) != true
+            }.ifEmpty { return@mapNotNull null }
+        }
+        deletedResults.forEach { (k, vs) ->
+            vs.forEach { v ->
+                _onValueRemoved.emit(k to v)
+            }
+        }
+        insertedResults.forEach { (k, vs) ->
+            vs.forEach { v ->
+                _onNewValue.emit(k to v)
+            }
+        }
     }
 
     override suspend fun remove(toRemove: Map<Key, List<Value>>) {