From 5755ab90b20ea21e4c46440117b26a9b1c976715 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 30 Apr 2026 15:48:36 +0000
Subject: [PATCH] refactor(quartz): switch project() builder from channelFlow
 to flow
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

channelFlow was overkill for a single-producer chain. The cost was
an extra Channel hop between every projection.snapshot() and the
downstream collector — pure overhead with no concurrency benefit.

flow { } needs the outer collector captured (because inside
changes.onSubscription { ... } and changes.collect { ... } the
implicit `this` is FlowCollector<StoreChange>, not
FlowCollector<ProjectionState<T>>). One `val outer = this` fixes
that.

Backpressure now flows directly: a slow collector suspends emit,
which suspends our changes.collect, which suspends the SharedFlow's
buffer drain — natural propagation, no decoupling Channel in the
middle.

17/17 projection tests pass.

https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
---
 .../cache/projection/EventStoreProjection.kt       | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt
index 60cccf113..bcd43f082 100644
--- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt
+++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt
@@ -35,7 +35,7 @@ import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent
 import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.channelFlow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.onSubscription
 import kotlinx.coroutines.yield
 
@@ -410,9 +410,13 @@ class EventStoreProjection<T : Event>(
  * NIP-62 vanish handling is scoped by the inner store's `relay`.
  */
 fun <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<ProjectionState<T>> =
-    channelFlow {
+    flow {
         val projection = EventStoreProjection<T>(this@project, filters)
-        send(ProjectionState.Loading)
+        // Capture the outer collector so we can emit ProjectionState
+        // from inside `changes.onSubscription { }` and `collect { }`,
+        // where the implicit `this` is FlowCollector<StoreChange>.
+        val outer = this
+        emit(ProjectionState.Loading)
         // `onSubscription` runs after the SharedFlow subscription is
         // active but before we pull events — the buffer absorbs
         // emissions arriving during the seed query and we drain them
@@ -422,9 +426,9 @@ fun <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<Projec
         changes
             .onSubscription {
                 projection.seed()
-                send(projection.snapshot())
+                outer.emit(projection.snapshot())
             }.collect { change ->
-                if (projection.apply(change)) send(projection.snapshot())
+                if (projection.apply(change)) outer.emit(projection.snapshot())
             }
     }