From f789fe4f53722506eec833302120a5e2d80fd205 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Apr 2026 15:35:30 +0000 Subject: [PATCH] refactor(quartz): EventStoreProjection becomes pure state machine; ObservableEventStore.project returns cold Flow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes scope ownership from EventStoreProjection. The class is now a pure state machine — no Job, no AutoCloseable, no embedded StateFlow: class EventStoreProjection(store, filters, nowProvider) { suspend fun seed() // run the seed query fun apply(change: StoreChange): Boolean // apply one mutation fun snapshot(): ProjectionState.Loaded // current view } ObservableEventStore.observe(filter, scope) is renamed to ObservableEventStore.project(filter) and now returns a cold Flow>. Each collection allocates its own state machine, runs its own seed, and unsubscribes when the collector cancels. The flow is built using channelFlow + the state machine — the class stays as the load-bearing primitive. Standard caller pattern: val state = observable.project(filter) .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), ProjectionState.Loading) Tradeoffs: - Cold flow → multiple collectors each re-seed. ViewModel pattern with stateIn shares one collection across observers (the standard Android approach). - close() goes away. Cancelling the collecting scope tears down the projection. - closeStopsListening test renamed to cancellingScopeStopsListening and now verifies the StateFlow's value freezes after its scope is cancelled. 17/17 projection + 240/240 store tests pass. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5 --- .../cache/projection/EventStoreProjection.kt | 255 +++++++++--------- .../projection/EventStoreProjectionTest.kt | 148 +++++----- 2 files changed, 185 insertions(+), 218 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 40e07a416..60cccf113 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 @@ -33,17 +33,14 @@ import com.vitorpamplona.quartz.nip40Expiration.isExpirationBefore import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent import com.vitorpamplona.quartz.utils.TimeUtils -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Job +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.channelFlow import kotlinx.coroutines.flow.onSubscription -import kotlinx.coroutines.launch import kotlinx.coroutines.yield /** - * Lifecycle state of an [EventStoreProjection]'s [state][EventStoreProjection.state]. + * Lifecycle state of an [ObservableEventStore.project] flow. * * - [Loading] is the initial state before the seed query completes. * The UI should show a spinner / skeleton here. @@ -60,31 +57,28 @@ sealed interface ProjectionState { } /** - * A reactive projection over an [ObservableEventStore] for a fixed - * set of [filters]. Each visible event is wrapped in a - * [MutableStateFlow] so the UI can collect three different kinds of - * change with the right granularity: + * State machine that maintains a reactive view over an + * [ObservableEventStore] for a fixed set of [filters]. Each visible + * event is wrapped in a [MutableStateFlow] so the UI can collect + * three different kinds of change with the right granularity: * - * - **Membership** (events arriving or leaving) re-emits a brand new - * [List] from [items]. The list reference is stable while membership - * is unchanged. + * - **Membership** (events arriving or leaving) produces a brand + * new [List] in [snapshot]. The list reference is stable while + * membership is unchanged. * - **In-place replaceable / addressable update** (a new version of - * the same `kind:pubkey:dtag` arrives) updates the existing handle's - * [MutableStateFlow.value] without touching the list. Only collectors - * of that one handle re-render. The list ordering is *not* - * reshuffled when the new version has a later `created_at` — each - * slot remembers the sort key it was inserted with, so updates feel - * like pure value mutations. + * the same `kind:pubkey:dtag` arrives) updates the existing + * handle's [MutableStateFlow.value] without changing membership. + * Only collectors of that one handle re-render. The list ordering + * is *not* reshuffled when the new version has a later + * `created_at` — each slot remembers the sort key it was inserted + * with, so updates feel like pure value mutations. * - **Removal** (NIP-09 deletion, NIP-62 vanish, NIP-40 expiration, * `delete(filter)`) drops the handle from the list. * - * The seed is materialised by querying the store once at start, after - * which the projection is driven entirely by - * [ObservableEventStore.changes]. Three kinds of mutation arrive on - * that stream: + * Three kinds of [StoreChange] are interpreted in-projection: * - * - [StoreChange.Insert] — interpreted in-projection so a single - * arriving event can carry NIP-01 / NIP-09 / NIP-62 semantics: + * - [StoreChange.Insert] — a single arriving event can carry NIP-01 + * / NIP-09 / NIP-62 semantics: * - **NIP-01 supersession.** New replaceable / addressable events * replace prior ones for the same `kind:pubkey[:dtag]`. The * NIP-01 lexical-id tiebreaker (`new.id < old.id` when @@ -97,55 +91,43 @@ sealed interface ProjectionState { * from the same author with `created_at < vanish.created_at`. * - **NIP-40 expiration.** Events whose `expiration` tag has * already lapsed at the moment they arrive are dropped before - * they ever enter [items]. - * + * they ever enter the snapshot. * - [StoreChange.DeleteByFilter] — emitted on `delete(filter)` / - * `delete(filters)`. The projection drops every slot matching any - * of the rule's filters via [Filter.match]. - * + * `delete(filters)`. Drops every slot matching any of the rule's + * filters via [Filter.match]. * - [StoreChange.DeleteExpired] — emitted on `deleteExpiredEvents()`. - * The projection drops every slot whose `expiration` has lapsed at - * the cutoff the store pinned. **There is no per-projection - * expiration ticker** — projections only drop expired events when - * the application calls `deleteExpiredEvents()` on the store. + * Drops every slot whose `expiration` has lapsed at the cutoff the + * store pinned. **There is no per-projection expiration ticker** — + * expiration only triggers when the application calls + * `deleteExpiredEvents()` on the store. * * Limit handling is **per-filter**: each filter retains at most its * own `limit` matches in a private capped set, sorted by created_at - * DESC + id ASC. The projection's [items] is the deduped union of - * those sets, so when filter A and filter B match disjoint events the - * union can be larger than any single filter's `limit`. We do not - * refill from the store after a deletion — if a removal leaves a - * filter under cap, it stays under cap until another match arrives. + * DESC + id ASC. The snapshot is the deduped union of those sets, so + * when filter A and filter B match disjoint events the union can be + * larger than any single filter's `limit`. We do not refill from the + * store after a deletion — if a removal leaves a filter under cap, it + * stays under cap until another match arrives. * - * Ephemeral events (kinds `20000-29999`) reach the projection via - * [ObservableEventStore.changes] without ever being persisted; they - * appear in [items] for as long as the projection is alive but never - * survive a re-seed. They aren't covered by the store's - * `deleteExpiredEvents()` sweep (the DB never had them), so an - * ephemeral with an `expiration` tag will linger in the projection + * Ephemeral events (kinds `20000-29999`) reach the projection without + * ever being persisted; they appear in the snapshot for as long as + * the projection is alive but never survive a re-seed. They aren't + * covered by the store's `deleteExpiredEvents()` sweep (the DB never + * had them), so an ephemeral with an `expiration` tag will linger * until it's superseded or until the projection is closed. * - * Lifecycle: the projection runs a single collector inside [scope]. - * Cancel the scope (or call [close]) when the screen using the - * projection goes away. + * **Lifecycle**: this class is a pure state machine with no + * coroutine ownership. Construct it, call [seed] once, then call + * [apply] for each [StoreChange] from [ObservableEventStore.changes]. + * For the standard "wrap into a Flow and collect from a ViewModel" + * use case, use the [ObservableEventStore.project] extension which + * does this composition for you. */ class EventStoreProjection( private val store: ObservableEventStore, private val filters: List, - scope: CoroutineScope, private val nowProvider: () -> Long = TimeUtils::now, -) : AutoCloseable { - private val _state = MutableStateFlow>(ProjectionState.Loading) - - /** - * Current projection state. Starts as [ProjectionState.Loading] - * until the seed query completes; thereafter every membership - * change publishes a fresh [ProjectionState.Loaded] with the new - * list. Distinguishing `Loading` from `Loaded(emptyList())` lets - * the UI tell "still seeding" apart from "seeded but no matches". - */ - val state: StateFlow> = _state.asStateFlow() - +) { /** Slots keyed by the *current* event id. Re-keyed when a replaceable / addressable handle takes a new version. */ private val byId = HashMap>() @@ -161,51 +143,56 @@ class EventStoreProjection( private val perFilter: Map>> = filters.associateWith { sortedSetOf(slotComparator()) } - private val collectorJob: Job = - scope.launch { - // `onSubscription` runs after the SharedFlow subscription is - // active but before we pull any events — the buffer absorbs - // emissions arriving during seed and drains them once collect - // proceeds. Doing the seed inside `collect { }` would race - // with concurrent inserts. - store.changes - .onSubscription { - seed() - }.collect { storeEvent -> apply(storeEvent) } - } - - private suspend fun seed() { + /** + * Run the seed query against the store and populate the indexes. + * Call once, before the first [apply]. Expired events are skipped + * (the store doesn't filter them at query time, so we do here). + */ + suspend fun seed() { val now = nowProvider() for (event in store.query(filters)) { - // Stores don't filter expired rows at query time, so do it - // here — otherwise an expired event would briefly appear - // in [items] before the next deleteExpiredEvents() sweep. if (event.isExpirationBefore(now)) continue applyInsert(event) yield() } - publish() } - private fun apply(storeEvent: StoreChange) { - val changed = - when (storeEvent) { - is StoreChange.Insert -> { - applyInsert(storeEvent.event) - } - - is StoreChange.DeleteByFilter -> { - dropWhere { ev -> storeEvent.filters.any { it.match(ev) } } - } - - // Store's sweep uses strict `<`; isExpirationBefore is - // `<=`, so subtract 1 to match. - is StoreChange.DeleteExpired -> { - val cutoff = (storeEvent.asOf ?: nowProvider()) - 1 - dropWhere { it.isExpirationBefore(cutoff) } - } + /** + * Apply a [StoreChange] from [ObservableEventStore.changes]. + * Returns true if the change altered membership (caller should + * publish a fresh [snapshot]); false for in-place addressable + * updates, NIP-01 tiebreaker rejections, and arrivals that no + * filter matches. + */ + fun apply(storeEvent: StoreChange): Boolean = + when (storeEvent) { + is StoreChange.Insert -> { + applyInsert(storeEvent.event) } - if (changed) publish() + + is StoreChange.DeleteByFilter -> { + dropWhere { ev -> storeEvent.filters.any { it.match(ev) } } + } + + // Store's sweep uses strict `<`; isExpirationBefore is + // `<=`, so subtract 1 to match. + is StoreChange.DeleteExpired -> { + val cutoff = (storeEvent.asOf ?: nowProvider()) - 1 + dropWhere { it.isExpirationBefore(cutoff) } + } + } + + /** + * Current snapshot as a [ProjectionState.Loaded]. Cheap — the + * deduped sorted union over [perFilter] sets is computed on + * demand. Returns an empty [ProjectionState.Loaded] when no slots + * are live. + */ + fun snapshot(): ProjectionState.Loaded { + if (byId.isEmpty()) return ProjectionState.Loaded(emptyList()) + val union = sortedSetOf(slotComparator()) + for (set in perFilter.values) union.addAll(set) + return ProjectionState.Loaded(union.map { it.flow }) } private fun applyInsert(event: Event): Boolean { @@ -341,32 +328,6 @@ class EventStoreProjection( return true } - private fun publish() { - // Lazily compute the deduped sorted union from per-filter - // sets. Cheaper than maintaining a separate `ordered` field - // alongside every insert / remove. - if (byId.isEmpty()) { - _state.value = ProjectionState.Loaded(emptyList()) - return - } - val union = sortedSetOf(slotComparator()) - for (set in perFilter.values) union.addAll(set) - _state.value = ProjectionState.Loaded(union.map { it.flow }) - } - - /** - * Stop tracking changes and clear internal state. Idempotent. The - * scope passed to the constructor keeps running; only this - * projection's collector job is cancelled. - */ - override fun close() { - collectorJob.cancel() - byId.clear() - byAddress.clear() - for (set in perFilter.values) set.clear() - _state.value = ProjectionState.Loaded(emptyList()) - } - /** * Internal slot. Each event added to the projection lives inside * one of these for as long as it survives. The sort key is frozen @@ -430,17 +391,41 @@ class EventStoreProjection( } /** - * Convenience: open a projection over this observable store. NIP-62 - * vanish handling is scoped by the inner store's `relay`. Cancel - * [scope] (or call [EventStoreProjection.close]) to release the - * projection. + * Open a cold reactive projection over this observable store for the + * given [filters]. The returned flow: + * + * - emits [ProjectionState.Loading] on subscription; + * - runs the seed query against the store; + * - emits [ProjectionState.Loaded] with the seeded list; + * - then emits a fresh [ProjectionState.Loaded] every time membership + * changes (in-place addressable updates do not re-emit — collectors + * of the slot's own [MutableStateFlow] handle those). + * + * Each collection allocates its own state machine, runs its own seed, + * and unsubscribes when the collector cancels — there's no + * `close()`. For multiple collectors on the same view, share with + * `stateIn(scope, SharingStarted.WhileSubscribed(...), Loading)` (the + * standard ViewModel pattern). + * + * NIP-62 vanish handling is scoped by the inner store's `relay`. */ -fun ObservableEventStore.observe( - filters: List, - scope: CoroutineScope, -): EventStoreProjection = EventStoreProjection(this, filters, scope) +fun ObservableEventStore.project(filters: List): Flow> = + channelFlow { + val projection = EventStoreProjection(this@project, filters) + send(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 + // through `apply` once collect proceeds. Doing the seed + // inside `collect { }` instead would race with concurrent + // inserts. + changes + .onSubscription { + projection.seed() + send(projection.snapshot()) + }.collect { change -> + if (projection.apply(change)) send(projection.snapshot()) + } + } -fun ObservableEventStore.observe( - filter: Filter, - scope: CoroutineScope, -): EventStoreProjection = EventStoreProjection(this, listOf(filter), scope) +fun ObservableEventStore.project(filter: Filter): Flow> = project(listOf(filter)) diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt index 4112b908b..68a98e28b 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt @@ -39,7 +39,10 @@ import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharingStarted +import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout import kotlin.test.AfterTest @@ -71,22 +74,27 @@ class EventStoreProjectionTest { store.close() } + /** Open a hot StateFlow over the projection for the lifetime of the test scope. */ + private fun projectionOf(filter: Filter): StateFlow> = observable.project(filter).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading) + + private fun projectionOf(filters: List): StateFlow> = observable.project(filters).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading) + /** Snapshot of the currently-loaded items, or empty if still seeding. */ - private val EventStoreProjection.items: List> - get() = (state.value as? ProjectionState.Loaded)?.items.orEmpty() + private val StateFlow>.items: List> + get() = (value as? ProjectionState.Loaded)?.items.orEmpty() /** Suspends until the seed completes; returns the loaded list. */ - private suspend fun EventStoreProjection.awaitReady(timeoutMs: Long = 5_000): List> = + private suspend fun StateFlow>.awaitLoaded(timeoutMs: Long = 5_000): List> = withTimeout(timeoutMs) { - (state.first { it is ProjectionState.Loaded } as ProjectionState.Loaded).items + (first { it is ProjectionState.Loaded } as ProjectionState.Loaded).items } - private suspend fun EventStoreProjection.awaitItems( + private suspend fun StateFlow>.awaitItems( timeoutMs: Long = 5_000, predicate: (List>) -> Boolean, ): List> = withTimeout(timeoutMs) { - (state.first { it is ProjectionState.Loaded && predicate(it.items) } as ProjectionState.Loaded).items + (first { it is ProjectionState.Loaded && predicate(it.items) } as ProjectionState.Loaded).items } private suspend fun awaitFlow( @@ -106,14 +114,13 @@ class EventStoreProjectionTest { observable.insert(a) observable.insert(b) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() val items = projection.items assertEquals(2, items.size) assertEquals(b.id, items[0].value.id) assertEquals(a.id, items[1].value.id) - projection.close() } @Test @@ -122,8 +129,8 @@ class EventStoreProjectionTest { val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) observable.insert(a) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() val before = projection.items assertEquals(1, before.size) @@ -133,7 +140,6 @@ class EventStoreProjectionTest { val after = projection.awaitItems { it.size == 2 } assertNotSame(before, after, "insert must produce a new list reference") assertEquals(b.id, after[0].value.id) - projection.close() } @Test @@ -142,8 +148,8 @@ class EventStoreProjectionTest { val text = signer.sign(TextNoteEvent.build("a", createdAt = 100)) observable.insert(text) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() val seed = projection.items val meta = signer.sign(MetadataEvent.createNew("Vitor", createdAt = 200)) @@ -151,7 +157,6 @@ class EventStoreProjectionTest { delay(150) assertSame(seed, projection.items) - projection.close() } @Test @@ -162,11 +167,8 @@ class EventStoreProjectionTest { observable.insert(v1) val projection = - observable.observe( - Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), - scope, - ) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey))) + projection.awaitLoaded() val seedList = projection.items assertEquals(1, seedList.size) val slot = seedList[0] @@ -178,7 +180,6 @@ class EventStoreProjectionTest { awaitFlow(slot) { it.id == v2.id } assertSame(seedList, projection.items, "replaceable update must not change list reference") assertSame(slot, projection.items[0]) - projection.close() } @Test @@ -189,15 +190,14 @@ class EventStoreProjectionTest { observable.insert(v1) val projection = - observable.observe( + projectionOf( Filter( kinds = listOf(LongTextNoteEvent.KIND), authors = listOf(v1.pubKey), tags = mapOf("d" to listOf("blog")), ), - scope, ) - projection.awaitReady() + projection.awaitLoaded() val seedList = projection.items val slot = seedList[0] @@ -206,7 +206,6 @@ class EventStoreProjectionTest { awaitFlow(slot) { it.id == v2.id } assertSame(seedList, projection.items, "addressable update must not change list reference") - projection.close() } /** @@ -228,11 +227,8 @@ class EventStoreProjectionTest { observable.insert(v2) val projection = - observable.observe( - Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), - scope, - ) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey))) + projection.awaitLoaded() val slot = projection.items[0] assertEquals(v2.id, slot.value.id) @@ -247,7 +243,6 @@ class EventStoreProjectionTest { delay(150) assertEquals(v2.id, slot.value.id) - projection.close() } @Test @@ -258,8 +253,8 @@ class EventStoreProjectionTest { observable.insert(a) observable.insert(b) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() assertEquals(2, projection.items.size) val deletion = signer.sign(DeletionEvent.build(listOf(a))) @@ -267,7 +262,6 @@ class EventStoreProjectionTest { val after = projection.awaitItems { it.size == 1 } assertEquals(b.id, after[0].value.id) - projection.close() } /** @@ -280,8 +274,8 @@ class EventStoreProjectionTest { val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) observable.insert(a) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() val seed = projection.items assertEquals(1, seed.size) @@ -296,7 +290,6 @@ class EventStoreProjectionTest { projection.items[0] .value.id, ) - projection.close() } @Test @@ -308,8 +301,8 @@ class EventStoreProjectionTest { observable.insert(a) observable.insert(b) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() assertEquals(2, projection.items.size) val vanish = @@ -323,7 +316,6 @@ class EventStoreProjectionTest { val after = projection.awaitItems { it.isEmpty() } assertTrue(after.isEmpty()) - projection.close() } /** @@ -337,8 +329,8 @@ class EventStoreProjectionTest { val a = signer.sign(TextNoteEvent.build("a", createdAt = time)) observable.insert(a) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() val seed = projection.items val foreignVanish = @@ -352,7 +344,6 @@ class EventStoreProjectionTest { delay(150) assertSame(seed, projection.items) - projection.close() } /** @@ -370,11 +361,8 @@ class EventStoreProjectionTest { observable.insert(short) val projection = - observable.observe( - Filter(kinds = listOf(TextNoteEvent.KIND)), - scope, - ) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() assertEquals(2, projection.items.size) // Let the short expiration lapse, then ask the store to @@ -385,7 +373,6 @@ class EventStoreProjectionTest { val after = projection.awaitItems { it.size == 1 } assertEquals(safe.id, after[0].value.id) - projection.close() } /** @@ -404,11 +391,8 @@ class EventStoreProjectionTest { observable.insert(foreign) val projection = - observable.observe( - Filter(kinds = listOf(TextNoteEvent.KIND)), - scope, - ) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() assertEquals(3, projection.items.size) // Drop everything authored by `signer` — should leave @@ -417,7 +401,6 @@ class EventStoreProjectionTest { val after = projection.awaitItems { it.size == 1 } assertEquals(foreign.id, after[0].value.id) - projection.close() } @Test @@ -429,11 +412,8 @@ class EventStoreProjectionTest { observable.insert(b) val projection = - observable.observe( - Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2), - scope, - ) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2)) + projection.awaitLoaded() assertEquals(2, projection.items.size) val c = signer.sign(TextNoteEvent.build("c", createdAt = 300)) @@ -443,7 +423,6 @@ class EventStoreProjectionTest { assertEquals(2, after.size) assertEquals(c.id, after[0].value.id) assertEquals(b.id, after[1].value.id) - projection.close() } /** @@ -468,13 +447,11 @@ class EventStoreProjectionTest { val filterA = Filter(kinds = listOf(TextNoteEvent.KIND), authors = listOf(authorA.pubKey), limit = 2) val filterB = Filter(kinds = listOf(TextNoteEvent.KIND), authors = listOf(authorB.pubKey), limit = 2) val projection = - observable.observe( + projectionOf( listOf(filterA, filterB), - scope, ) - projection.awaitReady() + projection.awaitLoaded() assertEquals(4, projection.items.size, "per-filter caps don't dedupe union") - projection.close() } /** @@ -491,11 +468,8 @@ class EventStoreProjectionTest { observable.insert(a2) val projection = - observable.observe( - Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2), - scope, - ) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2)) + projection.awaitLoaded() assertEquals(2, projection.items.size) observable.insert(a3) @@ -503,22 +477,33 @@ class EventStoreProjectionTest { assertEquals(2, after.size) assertEquals(a3.id, after[0].value.id) assertEquals(a2.id, after[1].value.id) - projection.close() } @Test - fun closeStopsListening() = + fun cancellingScopeStopsListening() = runBlocking { val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) observable.insert(a) - val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) - projection.awaitReady() - projection.close() + // Sub-scope so we can cancel just the projection's collector + // without taking down the test's outer scope. + val collectorScope = CoroutineScope(SupervisorJob()) + val projection = + observable + .project(Filter(kinds = listOf(TextNoteEvent.KIND))) + .stateIn(collectorScope, SharingStarted.Eagerly, ProjectionState.Loading) + projection.awaitLoaded() + assertEquals(1, projection.items.size) + collectorScope.cancel() + delay(50) + + // After the collector scope dies, new inserts don't update + // the StateFlow — its value is frozen at the last emission. observable.insert(signer.sign(TextNoteEvent.build("b", createdAt = 200))) delay(150) - assertTrue(projection.items.isEmpty()) + assertEquals(1, projection.items.size) + assertEquals(a.id, projection.items[0].value.id) } /** @@ -534,8 +519,8 @@ class EventStoreProjectionTest { runBlocking { val ephemeralKind = 22_000 val projection = - observable.observe(Filter(kinds = listOf(ephemeralKind)), scope) - projection.awaitReady() + projectionOf(Filter(kinds = listOf(ephemeralKind))) + projection.awaitLoaded() assertTrue(projection.items.isEmpty()) val ephemeral: Event = @@ -557,11 +542,8 @@ class EventStoreProjectionTest { // A fresh projection on the same store gets nothing — the // event was only ever live, not durable. val freshProjection = - observable.observe(Filter(kinds = listOf(ephemeralKind)), scope) - freshProjection.awaitReady() + projectionOf(Filter(kinds = listOf(ephemeralKind))) + freshProjection.awaitLoaded() assertTrue(freshProjection.items.isEmpty()) - - projection.close() - freshProjection.close() } }