Merge pull request #138 from soil-kt/introduce-subscription-filters

Introduce Subscription Filters

internal/playground/src/commonMain/kotlin/soil/playground/query/key/posts/CreatePostKey.kt

@@ -8,9 +8,10 @@ import soil.playground.query.data.Post
 import soil.query.InfiniteQueryId
 import soil.query.MutationId
 import soil.query.MutationKey
-import soil.query.QueryEffect
+import soil.query.core.Effect
 import soil.query.core.KeyEquals
 import soil.query.core.Namespace
+import soil.query.queryClient
 import soil.query.receivers.ktor.buildKtorMutationKey
 
 @Stable
@@ -22,8 +23,9 @@ class CreatePostKey(auto: Namespace) : KeyEquals(), MutationKey<Post, PostForm>
         }.body()
     }
 ) {
-    override fun onQueryUpdate(variable: PostForm, data: Post): QueryEffect = {
-        invalidateQueriesBy(InfiniteQueryId.forGetPosts())
+
+    override fun onMutateEffect(variable: PostForm, data: Post): Effect = {
+        queryClient.invalidateQueriesBy(InfiniteQueryId.forGetPosts())
     }
 }
 

internal/playground/src/commonMain/kotlin/soil/playground/query/key/posts/DeletePostKey.kt

@@ -5,11 +5,12 @@ import io.ktor.client.request.delete
 import soil.query.InfiniteQueryId
 import soil.query.MutationId
 import soil.query.MutationKey
-import soil.query.QueryEffect
 import soil.query.QueryId
+import soil.query.core.Effect
 import soil.query.core.KeyEquals
 import soil.query.core.Namespace
 import soil.query.receivers.ktor.buildKtorMutationKey
+import soil.query.withQuery
 
 @Stable
 class DeletePostKey(auto: Namespace) : KeyEquals(), MutationKey<Unit, Int> by buildKtorMutationKey(
@@ -18,8 +19,10 @@ class DeletePostKey(auto: Namespace) : KeyEquals(), MutationKey<Unit, Int> by bu
         delete("https://jsonplaceholder.typicode.com/posts/$postId")
     }
 ) {
-    override fun onQueryUpdate(variable: Int, data: Unit): QueryEffect = {
-        removeQueriesBy(QueryId.forGetPost(variable))
-        invalidateQueriesBy(InfiniteQueryId.forGetPosts())
+    override fun onMutateEffect(variable: Int, data: Unit): Effect = {
+        withQuery {
+            removeQueriesBy(QueryId.forGetPost(variable))
+            invalidateQueriesBy(InfiniteQueryId.forGetPosts())
+        }
     }
 }

internal/playground/src/commonMain/kotlin/soil/playground/query/key/posts/UpdatePostKey.kt

@@ -8,12 +8,13 @@ import soil.playground.query.data.Post
 import soil.query.InfiniteQueryId
 import soil.query.MutationId
 import soil.query.MutationKey
-import soil.query.QueryEffect
 import soil.query.QueryId
+import soil.query.core.Effect
 import soil.query.core.KeyEquals
 import soil.query.core.Namespace
 import soil.query.modifyData
 import soil.query.receivers.ktor.buildKtorMutationKey
+import soil.query.withQuery
 
 @Stable
 class UpdatePostKey(auto: Namespace) : KeyEquals(), MutationKey<Post, Post> by buildKtorMutationKey(
@@ -24,8 +25,11 @@ class UpdatePostKey(auto: Namespace) : KeyEquals(), MutationKey<Post, Post> by b
         }.body()
     }
 ) {
-    override fun onQueryUpdate(variable: Post, data: Post): QueryEffect = {
-        updateQueryData(QueryId.forGetPost(data.id)) { data }
-        updateInfiniteQueryData(InfiniteQueryId.forGetPosts()) { modifyData({ it.id == data.id }) { data } }
+
+    override fun onMutateEffect(variable: Post, data: Post): Effect = {
+        withQuery {
+            updateQueryData(QueryId.forGetPost(data.id)) { data }
+            updateInfiniteQueryData(InfiniteQueryId.forGetPosts()) { modifyData({ it.id == data.id }) { data } }
+        }
     }
 }

soil-query-compose/src/commonMain/kotlin/soil/query/compose/SubscriptionComposable.kt

@@ -4,10 +4,12 @@
 package soil.query.compose
 
 import androidx.compose.runtime.Composable
+import androidx.compose.runtime.LaunchedEffect
 import androidx.compose.runtime.remember
 import androidx.compose.runtime.rememberCoroutineScope
 import soil.query.SubscriptionClient
 import soil.query.SubscriptionKey
+import soil.query.SubscriptionRef
 import soil.query.annotation.ExperimentalSoilQueryApi
 import soil.query.compose.internal.newSubscription
 
@@ -29,6 +31,7 @@ fun <T> rememberSubscription(
 ): SubscriptionObject<T> {
     val scope = rememberCoroutineScope()
     val subscription = remember(key.id) { newSubscription(key, config, client, scope) }
+    subscription.Effect()
     return with(config.mapper) {
         config.strategy.collectAsState(subscription).toObject(subscription = subscription, select = { it })
     }
@@ -55,7 +58,19 @@ fun <T, U> rememberSubscription(
 ): SubscriptionObject<U> {
     val scope = rememberCoroutineScope()
     val subscription = remember(key.id) { newSubscription(key, config, client, scope) }
+    subscription.Effect()
     return with(config.mapper) {
         config.strategy.collectAsState(subscription).toObject(subscription = subscription, select = select)
     }
 }
+
+@Suppress("NOTHING_TO_INLINE", "KotlinRedundantDiagnosticSuppress")
+@Composable
+private inline fun SubscriptionRef<*>.Effect() {
+    // TODO: Switch to LifecycleResumeEffect
+    //  Android, it works only with Compose UI 1.7.0-alpha05 or above.
+    //  Therefore, we will postpone adding this code until a future release.
+    LaunchedEffect(id) {
+        join()
+    }
+}

soil-query-compose/src/commonMain/kotlin/soil/query/compose/SubscriptionObject.kt

@@ -43,6 +43,7 @@ data class SubscriptionLoadingObject<T>(
     override val replyUpdatedAt: Long,
     override val error: Throwable?,
     override val errorUpdatedAt: Long,
+    override val restartedAt: Long,
     override val reset: suspend () -> Unit
 ) : SubscriptionObject<T> {
     override val status: SubscriptionStatus = SubscriptionStatus.Pending
@@ -63,6 +64,7 @@ data class SubscriptionErrorObject<T>(
     override val replyUpdatedAt: Long,
     override val error: Throwable,
     override val errorUpdatedAt: Long,
+    override val restartedAt: Long,
     override val reset: suspend () -> Unit
 ) : SubscriptionObject<T> {
     override val status: SubscriptionStatus = SubscriptionStatus.Failure
@@ -80,6 +82,7 @@ data class SubscriptionSuccessObject<T>(
     override val replyUpdatedAt: Long,
     override val error: Throwable?,
     override val errorUpdatedAt: Long,
+    override val restartedAt: Long,
     override val reset: suspend () -> Unit
 ) : SubscriptionObject<T> {
     override val status: SubscriptionStatus = SubscriptionStatus.Success

soil-query-compose/src/commonMain/kotlin/soil/query/compose/SubscriptionObjectMapper.kt

@@ -44,6 +44,7 @@ private object DefaultSubscriptionObjectMapper : SubscriptionObjectMapper {
             replyUpdatedAt = replyUpdatedAt,
             error = error,
             errorUpdatedAt = errorUpdatedAt,
+            restartedAt = restartedAt,
             reset = subscription::reset
         )
 
@@ -52,6 +53,7 @@ private object DefaultSubscriptionObjectMapper : SubscriptionObjectMapper {
             replyUpdatedAt = replyUpdatedAt,
             error = error,
             errorUpdatedAt = errorUpdatedAt,
+            restartedAt = restartedAt,
             reset = subscription::reset
         )
 
@@ -60,6 +62,7 @@ private object DefaultSubscriptionObjectMapper : SubscriptionObjectMapper {
             replyUpdatedAt = replyUpdatedAt,
             error = checkNotNull(error),
             errorUpdatedAt = errorUpdatedAt,
+            restartedAt = restartedAt,
             reset = subscription::reset
         )
     }

soil-query-compose/src/commonMain/kotlin/soil/query/compose/SubscriptionRecompositionOptimizer.kt

@@ -32,6 +32,7 @@ private object SubscriptionRecompositionOptimizerEnabled : SubscriptionRecomposi
     override fun <T> omit(state: SubscriptionState<T>): SubscriptionState<T> {
         val keys = buildSet {
             add(SubscriptionState.OmitKey.replyUpdatedAt)
+            add(SubscriptionState.OmitKey.restartedAt)
             when (state.status) {
                 SubscriptionStatus.Pending -> {
                     add(SubscriptionState.OmitKey.errorUpdatedAt)

soil-query-compose/src/commonMain/kotlin/soil/query/compose/internal/Subscription.kt

@@ -49,6 +49,8 @@ private class Subscription<T>(
 
     override suspend fun resume() = subscription.resume()
 
+    override suspend fun join() = subscription.join()
+
     // ----- RememberObserver -----//
     private var job: Job? = null
 

soil-query-compose/src/commonMain/kotlin/soil/query/compose/tooling/SubscriptionPreviewClient.kt

@@ -46,6 +46,7 @@ class SubscriptionPreviewClient(
         override fun close() = Unit
         override suspend fun reset() = Unit
         override suspend fun resume() = Unit
+        override suspend fun join() = Unit
     }
 
     /**

soil-query-compose/src/commonMain/kotlin/soil/query/compose/tooling/SwrPreviewClient.kt

@@ -13,6 +13,7 @@ import soil.query.QueryEffect
 import soil.query.SubscriptionClient
 import soil.query.SwrClient
 import soil.query.SwrClientPlus
+import soil.query.core.Effect
 import soil.query.core.ErrorRecord
 import soil.query.core.MemoryPressureLevel
 
@@ -35,7 +36,10 @@ class SwrPreviewClient(
 ) : SwrClient, SwrClientPlus, QueryClient by query, MutationClient by mutation, SubscriptionClient by subscription {
     override fun gc(level: MemoryPressureLevel) = Unit
     override fun purgeAll() = Unit
+
+    @Deprecated("Use effect(block: Effect) instead.", replaceWith = ReplaceWith("effect(block)"))
     override fun perform(sideEffects: QueryEffect): Job = Job()
+    override fun effect(block: Effect): Job = Job()
     override fun onMount(id: String) = Unit
     override fun onUnmount(id: String) = Unit
 }

soil-query-compose/src/commonMain/kotlin/soil/query/compose/util/QueriesErrorReset.kt

@@ -8,6 +8,7 @@ import androidx.compose.runtime.remember
 import soil.query.ResumeQueriesFilter
 import soil.query.SwrClient
 import soil.query.compose.LocalSwrClient
+import soil.query.queryClient
 
 
 typealias QueriesErrorReset = () -> Unit
@@ -25,7 +26,7 @@ fun rememberQueriesErrorReset(
     client: SwrClient = LocalSwrClient.current
 ): QueriesErrorReset {
     val reset = remember<() -> Unit>(client) {
-        { client.perform { resumeQueries(filter) } }
+        { client.effect { queryClient.resumeQueries(filter) } }
     }
     return reset
 }

soil-query-core/src/commonMain/kotlin/soil/query/MutationCommand.kt

@@ -99,10 +99,16 @@ suspend inline fun <T, S> MutationCommand.Context<T>.dispatchMutateResult(
 ) {
     mutate(key, variable)
         .onSuccess { data ->
-            val job = key.onQueryUpdate(variable, data)?.let(notifier::onMutate)
+            val job1 = key.onQueryUpdate(variable, data)?.let {
+                notifier.onMutate { withQuery { it() } }
+            }
+            val job2 = key.onMutateEffect(variable, data)?.let(notifier::onMutate)
             withContext(NonCancellable) {
-                if (job != null && options.shouldExecuteEffectSynchronously) {
-                    job.join()
+                if (job1 != null && options.shouldExecuteEffectSynchronously) {
+                    job1.join()
+                }
+                if (job2 != null && options.shouldExecuteEffectSynchronously) {
+                    job2.join()
                 }
                 dispatchMutateSuccess(data, key.contentEquals)
             }

soil-query-core/src/commonMain/kotlin/soil/query/MutationKey.kt

@@ -3,6 +3,7 @@
 
 package soil.query
 
+import soil.query.core.Effect
 import soil.query.core.SurrogateKey
 import soil.query.core.UniqueId
 import soil.query.core.uuid
@@ -70,7 +71,27 @@ interface MutationKey<T, S> {
      * @param variable The variable to be mutated.
      * @param data The data returned by the mutation.
      */
+    @Deprecated(
+        "Use onMutateEffect instead.",
+        ReplaceWith("onMutateEffect(variable, data)")
+    )
     fun onQueryUpdate(variable: S, data: T): QueryEffect? = null
+
+    /**
+     * Function to handle side effects after the mutation is executed.
+     *
+     * This is often referred to as ["Pessimistic Updates"](https://redux-toolkit.js.org/rtk-query/usage/manual-cache-updates#pessimistic-updates).
+     *
+     * ```kotlin
+     * override fun onMutateEffect(variable: PostForm, data: Post): Effect = {
+     *     queryClient.invalidateQueriesBy(GetPostsKey.Id())
+     * }
+     * ```
+     *
+     * @param variable The variable to be mutated.
+     * @param data The data returned by the mutation.
+     */
+    fun onMutateEffect(variable: S, data: T): Effect? = null
 }
 
 /**

soil-query-core/src/commonMain/kotlin/soil/query/MutationNotifier.kt

@@ -4,6 +4,7 @@
 package soil.query
 
 import kotlinx.coroutines.Job
+import soil.query.core.Effect
 
 /**
  * MutationNotifier is used to notify the mutation result.
@@ -13,11 +14,11 @@ fun interface MutationNotifier {
     /**
      * Notifies the mutation success
      *
-     * Mutation success usually implies data update, causing side effects on related queries.
+     * Mutation success usually implies data update, causing side effects on related queries and subscriptions.
      * This callback is used as a trigger for re-fetching or revalidating data managed by queries.
-     * It invokes with the [QueryEffect] set in [MutationKey.onQueryUpdate].
+     * It invokes with the [Effect] set in [MutationKey.onMutateEffect].
      *
-     * @param sideEffects The side effects of the mutation for related queries.
+     * @param effect The side effects of the mutation for related queries and subscriptions.
      */
-    fun onMutate(sideEffects: QueryEffect): Job
+    fun onMutate(effect: Effect): Job
 }

soil-query-core/src/commonMain/kotlin/soil/query/QueryClient.kt

@@ -5,7 +5,6 @@ package soil.query
 
 import kotlinx.coroutines.Job
 import soil.query.core.Marker
-import soil.query.core.UniqueId
 
 /**
  * A Query client, which allows you to make queries actor and handle [QueryKey] and [InfiniteQueryKey].
@@ -72,68 +71,7 @@ interface QueryReadonlyClient {
     fun <T, S> getInfiniteQueryData(id: InfiniteQueryId<T, S>): QueryChunks<T, S>?
 }
 
-/**
- * Interface for causing side effects on [Query] under the control of [QueryClient].
- *
- * [QueryEffect] is designed to allow side effects such as updating, deleting, and revalidating queries.
- * It is useful for handling [MutationKey.onQueryUpdate] after executing [Mutation] that affects [Query] data.
- */
-interface QueryMutableClient : QueryReadonlyClient {
-
-    /**
-     * Updates the data of the [QueryKey] associated with the [id].
-     */
-    fun <T> updateQueryData(
-        id: QueryId<T>,
-        edit: T.() -> T
-    )
-
-    /**
-     * Updates the data of the [InfiniteQueryKey] associated with the [id].
-     */
-    fun <T, S> updateInfiniteQueryData(
-        id: InfiniteQueryId<T, S>,
-        edit: QueryChunks<T, S>.() -> QueryChunks<T, S>
-    )
-
-    /**
-     * Invalidates the queries by the specified [InvalidateQueriesFilter].
-     */
-    fun invalidateQueries(filter: InvalidateQueriesFilter)
-
-    /**
-     * Invalidates the queries by the specified [UniqueId].
-     */
-    fun <U : UniqueId> invalidateQueriesBy(vararg ids: U)
-
-    /**
-     * Removes the queries by the specified [RemoveQueriesFilter].
-     *
-     * **Note:**
-     * Queries will be removed from [QueryClient], but [QueryRef] instances on the subscriber side will remain until they are dereferenced.
-     * Also, the [kotlinx.coroutines.CoroutineScope] associated with the [kotlinx.coroutines.Job] will be canceled at the time of removal.
-     */
-    fun removeQueries(filter: RemoveQueriesFilter)
-
-    /**
-     * Removes the queries by the specified [UniqueId].
-     */
-    fun <U : UniqueId> removeQueriesBy(vararg ids: U)
-
-    /**
-     * Resumes the queries by the specified [ResumeQueriesFilter].
-     */
-    fun resumeQueries(filter: ResumeQueriesFilter)
-
-    /**
-     * Resumes the queries by the specified [UniqueId].
-     */
-    fun <U : UniqueId> resumeQueriesBy(vararg ids: U)
-}
-
 typealias QueryInitialData<T> = QueryReadonlyClient.() -> T?
-typealias QueryEffect = QueryMutableClient.() -> Unit
-
 typealias QueryContentEquals<T> = (oldData: T, newData: T) -> Boolean
 typealias QueryContentCacheable<T> = (currentData: T) -> Boolean
 typealias QueryRecoverData<T> = (error: Throwable) -> T