nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(quartz): EventStoreProjection becomes pure state machine; ObservableEventStore.project returns cold Flow

Browse Source 
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<ProjectionState<T>>. 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<TextNoteEvent>(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
This commit is contained in:
Claude
2026-04-30 15:35:30 +00:00
parent 8517bae7a4
commit f789fe4f53
2 changed files with 185 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt
Vendored
+105 -120
View File
@@ -33,17 +33,14 @@ import com.vitorpamplona.quartz.nip40Expiration.isExpirationBefore
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent
import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.TimeUtils
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.onSubscription import kotlinx.coroutines.flow.onSubscription
import kotlinx.coroutines.launch
import kotlinx.coroutines.yield 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. * - [Loading] is the initial state before the seed query completes.
* The UI should show a spinner / skeleton here. * The UI should show a spinner / skeleton here.
@@ -60,31 +57,28 @@ sealed interface ProjectionState<out T : Event> {
} }
/** /**
* A reactive projection over an [ObservableEventStore] for a fixed * State machine that maintains a reactive view over an
* set of [filters]. Each visible event is wrapped in a * [ObservableEventStore] for a fixed set of [filters]. Each visible
* [MutableStateFlow] so the UI can collect three different kinds of * event is wrapped in a [MutableStateFlow] so the UI can collect
* change with the right granularity: * three different kinds of change with the right granularity:
* *
* - **Membership** (events arriving or leaving) re-emits a brand new * - **Membership** (events arriving or leaving) produces a brand
* [List] from [items]. The list reference is stable while membership * new [List] in [snapshot]. The list reference is stable while
* is unchanged. * membership is unchanged.
* - **In-place replaceable / addressable update** (a new version of * - **In-place replaceable / addressable update** (a new version of
* the same `kind:pubkey:dtag` arrives) updates the existing handle's * the same `kind:pubkey:dtag` arrives) updates the existing
* [MutableStateFlow.value] without touching the list. Only collectors * handle's [MutableStateFlow.value] without changing membership.
* of that one handle re-render. The list ordering is *not* * Only collectors of that one handle re-render. The list ordering
* reshuffled when the new version has a later `created_at` — each * is *not* reshuffled when the new version has a later
* slot remembers the sort key it was inserted with, so updates feel * `created_at` — each slot remembers the sort key it was inserted
* like pure value mutations. * with, so updates feel like pure value mutations.
* - **Removal** (NIP-09 deletion, NIP-62 vanish, NIP-40 expiration, * - **Removal** (NIP-09 deletion, NIP-62 vanish, NIP-40 expiration,
* `delete(filter)`) drops the handle from the list. * `delete(filter)`) drops the handle from the list.
* *
* The seed is materialised by querying the store once at start, after * Three kinds of [StoreChange] are interpreted in-projection:
* which the projection is driven entirely by
* [ObservableEventStore.changes]. Three kinds of mutation arrive on
* that stream:
* *
* - [StoreChange.Insert] — interpreted in-projection so a single * - [StoreChange.Insert] — a single arriving event can carry NIP-01
* arriving event can carry NIP-01 / NIP-09 / NIP-62 semantics: * / NIP-09 / NIP-62 semantics:
* - **NIP-01 supersession.** New replaceable / addressable events * - **NIP-01 supersession.** New replaceable / addressable events
* replace prior ones for the same `kind:pubkey[:dtag]`. The * replace prior ones for the same `kind:pubkey[:dtag]`. The
* NIP-01 lexical-id tiebreaker (`new.id < old.id` when * NIP-01 lexical-id tiebreaker (`new.id < old.id` when
@@ -97,55 +91,43 @@ sealed interface ProjectionState<out T : Event> {
* from the same author with `created_at < vanish.created_at`. * from the same author with `created_at < vanish.created_at`.
* - **NIP-40 expiration.** Events whose `expiration` tag has * - **NIP-40 expiration.** Events whose `expiration` tag has
* already lapsed at the moment they arrive are dropped before * 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)` / * - [StoreChange.DeleteByFilter] — emitted on `delete(filter)` /
* `delete(filters)`. The projection drops every slot matching any * `delete(filters)`. Drops every slot matching any of the rule's
* of the rule's filters via [Filter.match]. * filters via [Filter.match].
*
* - [StoreChange.DeleteExpired] — emitted on `deleteExpiredEvents()`. * - [StoreChange.DeleteExpired] — emitted on `deleteExpiredEvents()`.
* The projection drops every slot whose `expiration` has lapsed at * Drops every slot whose `expiration` has lapsed at the cutoff the
* the cutoff the store pinned. **There is no per-projection * store pinned. **There is no per-projection expiration ticker** —
* expiration ticker** — projections only drop expired events when * expiration only triggers when the application calls
* the application calls `deleteExpiredEvents()` on the store. * `deleteExpiredEvents()` on the store.
* *
* Limit handling is **per-filter**: each filter retains at most its * Limit handling is **per-filter**: each filter retains at most its
* own `limit` matches in a private capped set, sorted by created_at * own `limit` matches in a private capped set, sorted by created_at
* DESC + id ASC. The projection's [items] is the deduped union of * DESC + id ASC. The snapshot is the deduped union of those sets, so
* those sets, so when filter A and filter B match disjoint events the * when filter A and filter B match disjoint events the union can be
* union can be larger than any single filter's `limit`. We do not * larger than any single filter's `limit`. We do not refill from the
* refill from the store after a deletion — if a removal leaves a * store after a deletion — if a removal leaves a filter under cap, it
* filter under cap, it stays under cap until another match arrives. * stays under cap until another match arrives.
* *
* Ephemeral events (kinds `20000-29999`) reach the projection via * Ephemeral events (kinds `20000-29999`) reach the projection without
* [ObservableEventStore.changes] without ever being persisted; they * ever being persisted; they appear in the snapshot for as long as
* appear in [items] for as long as the projection is alive but never * the projection is alive but never survive a re-seed. They aren't
* survive a re-seed. They aren't covered by the store's * covered by the store's `deleteExpiredEvents()` sweep (the DB never
* `deleteExpiredEvents()` sweep (the DB never had them), so an * had them), so an ephemeral with an `expiration` tag will linger
* ephemeral with an `expiration` tag will linger in the projection
* until it's superseded or until the projection is closed. * until it's superseded or until the projection is closed.
* *
* Lifecycle: the projection runs a single collector inside [scope]. * **Lifecycle**: this class is a pure state machine with no
* Cancel the scope (or call [close]) when the screen using the * coroutine ownership. Construct it, call [seed] once, then call
* projection goes away. * [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<T : Event>( class EventStoreProjection<T : Event>(
private val store: ObservableEventStore, private val store: ObservableEventStore,
private val filters: List<Filter>, private val filters: List<Filter>,
scope: CoroutineScope,
private val nowProvider: () -> Long = TimeUtils::now, private val nowProvider: () -> Long = TimeUtils::now,
) : AutoCloseable { ) {
private val _state = MutableStateFlow<ProjectionState<T>>(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<ProjectionState<T>> = _state.asStateFlow()
/** Slots keyed by the *current* event id. Re-keyed when a replaceable / addressable handle takes a new version. */ /** Slots keyed by the *current* event id. Re-keyed when a replaceable / addressable handle takes a new version. */
private val byId = HashMap<HexKey, Slot<T>>() private val byId = HashMap<HexKey, Slot<T>>()
@@ -161,34 +143,28 @@ class EventStoreProjection<T : Event>(
private val perFilter: Map<Filter, java.util.SortedSet<Slot<T>>> = private val perFilter: Map<Filter, java.util.SortedSet<Slot<T>>> =
filters.associateWith { sortedSetOf(slotComparator()) } filters.associateWith { sortedSetOf(slotComparator()) }
private val collectorJob: Job = /**
scope.launch { * Run the seed query against the store and populate the indexes.
// `onSubscription` runs after the SharedFlow subscription is * Call once, before the first [apply]. Expired events are skipped
// active but before we pull any events — the buffer absorbs * (the store doesn't filter them at query time, so we do here).
// emissions arriving during seed and drains them once collect */
// proceeds. Doing the seed inside `collect { }` would race suspend fun seed() {
// with concurrent inserts.
store.changes
.onSubscription {
seed()
}.collect { storeEvent -> apply(storeEvent) }
}
private suspend fun seed() {
val now = nowProvider() val now = nowProvider()
for (event in store.query<T>(filters)) { for (event in store.query<T>(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 if (event.isExpirationBefore(now)) continue
applyInsert(event) applyInsert(event)
yield() yield()
} }
publish()
} }
private fun apply(storeEvent: StoreChange) { /**
val changed = * 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) { when (storeEvent) {
is StoreChange.Insert -> { is StoreChange.Insert -> {
applyInsert(storeEvent.event) applyInsert(storeEvent.event)
@@ -205,7 +181,18 @@ class EventStoreProjection<T : Event>(
dropWhere { it.isExpirationBefore(cutoff) } dropWhere { it.isExpirationBefore(cutoff) }
} }
} }
if (changed) publish()
/**
* 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<T> {
if (byId.isEmpty()) return ProjectionState.Loaded(emptyList())
val union = sortedSetOf(slotComparator<T>())
for (set in perFilter.values) union.addAll(set)
return ProjectionState.Loaded(union.map { it.flow })
} }
private fun applyInsert(event: Event): Boolean { private fun applyInsert(event: Event): Boolean {
@@ -341,32 +328,6 @@ class EventStoreProjection<T : Event>(
return true 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<T>())
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 * Internal slot. Each event added to the projection lives inside
* one of these for as long as it survives. The sort key is frozen * one of these for as long as it survives. The sort key is frozen
@@ -430,17 +391,41 @@ class EventStoreProjection<T : Event>(
} }
/** /**
* Convenience: open a projection over this observable store. NIP-62 * Open a cold reactive projection over this observable store for the
* vanish handling is scoped by the inner store's `relay`. Cancel * given [filters]. The returned flow:
* [scope] (or call [EventStoreProjection.close]) to release the *
* projection. * - 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 <T : Event> ObservableEventStore.observe( fun <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<ProjectionState<T>> =
filters: List<Filter>, channelFlow {
scope: CoroutineScope, val projection = EventStoreProjection<T>(this@project, filters)
): EventStoreProjection<T> = EventStoreProjection(this, filters, scope) 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 <T : Event> ObservableEventStore.observe( fun <T : Event> ObservableEventStore.project(filter: Filter): Flow<ProjectionState<T>> = project(listOf(filter))
filter: Filter,
scope: CoroutineScope,
): EventStoreProjection<T> = EventStoreProjection(this, listOf(filter), scope)
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt
Vendored
+65 -83
View File
@@ -39,7 +39,10 @@ import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout import kotlinx.coroutines.withTimeout
import kotlin.test.AfterTest import kotlin.test.AfterTest
@@ -71,22 +74,27 @@ class EventStoreProjectionTest {
store.close() store.close()
} }
/** Open a hot StateFlow over the projection for the lifetime of the test scope. */
private fun <T : Event> projectionOf(filter: Filter): StateFlow<ProjectionState<T>> = observable.project<T>(filter).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading)
private fun <T : Event> projectionOf(filters: List<Filter>): StateFlow<ProjectionState<T>> = observable.project<T>(filters).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading)
/** Snapshot of the currently-loaded items, or empty if still seeding. */ /** Snapshot of the currently-loaded items, or empty if still seeding. */
private val <T : Event> EventStoreProjection<T>.items: List<MutableStateFlow<T>> private val <T : Event> StateFlow<ProjectionState<T>>.items: List<MutableStateFlow<T>>
get() = (state.value as? ProjectionState.Loaded)?.items.orEmpty() get() = (value as? ProjectionState.Loaded)?.items.orEmpty()
/** Suspends until the seed completes; returns the loaded list. */ /** Suspends until the seed completes; returns the loaded list. */
private suspend fun <T : Event> EventStoreProjection<T>.awaitReady(timeoutMs: Long = 5_000): List<MutableStateFlow<T>> = private suspend fun <T : Event> StateFlow<ProjectionState<T>>.awaitLoaded(timeoutMs: Long = 5_000): List<MutableStateFlow<T>> =
withTimeout(timeoutMs) { withTimeout(timeoutMs) {
(state.first { it is ProjectionState.Loaded } as ProjectionState.Loaded).items (first { it is ProjectionState.Loaded } as ProjectionState.Loaded).items
} }
private suspend fun <T : Event> EventStoreProjection<T>.awaitItems( private suspend fun <T : Event> StateFlow<ProjectionState<T>>.awaitItems(
timeoutMs: Long = 5_000, timeoutMs: Long = 5_000,
predicate: (List<MutableStateFlow<T>>) -> Boolean, predicate: (List<MutableStateFlow<T>>) -> Boolean,
): List<MutableStateFlow<T>> = ): List<MutableStateFlow<T>> =
withTimeout(timeoutMs) { 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 <T : Event> awaitFlow( private suspend fun <T : Event> awaitFlow(
@@ -106,14 +114,13 @@ class EventStoreProjectionTest {
observable.insert(a) observable.insert(a)
observable.insert(b) observable.insert(b)
val projection = observable.observe<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
val items = projection.items val items = projection.items
assertEquals(2, items.size) assertEquals(2, items.size)
assertEquals(b.id, items[0].value.id) assertEquals(b.id, items[0].value.id)
assertEquals(a.id, items[1].value.id) assertEquals(a.id, items[1].value.id)
projection.close()
} }
@Test @Test
@@ -122,8 +129,8 @@ class EventStoreProjectionTest {
val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) val a = signer.sign(TextNoteEvent.build("a", createdAt = 100))
observable.insert(a) observable.insert(a)
val projection = observable.observe<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
val before = projection.items val before = projection.items
assertEquals(1, before.size) assertEquals(1, before.size)
@@ -133,7 +140,6 @@ class EventStoreProjectionTest {
val after = projection.awaitItems { it.size == 2 } val after = projection.awaitItems { it.size == 2 }
assertNotSame(before, after, "insert must produce a new list reference") assertNotSame(before, after, "insert must produce a new list reference")
assertEquals(b.id, after[0].value.id) assertEquals(b.id, after[0].value.id)
projection.close()
} }
@Test @Test
@@ -142,8 +148,8 @@ class EventStoreProjectionTest {
val text = signer.sign(TextNoteEvent.build("a", createdAt = 100)) val text = signer.sign(TextNoteEvent.build("a", createdAt = 100))
observable.insert(text) observable.insert(text)
val projection = observable.observe<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
val seed = projection.items val seed = projection.items
val meta = signer.sign(MetadataEvent.createNew("Vitor", createdAt = 200)) val meta = signer.sign(MetadataEvent.createNew("Vitor", createdAt = 200))
@@ -151,7 +157,6 @@ class EventStoreProjectionTest {
delay(150) delay(150)
assertSame(seed, projection.items) assertSame(seed, projection.items)
projection.close()
} }
@Test @Test
@@ -162,11 +167,8 @@ class EventStoreProjectionTest {
observable.insert(v1) observable.insert(v1)
val projection = val projection =
observable.observe<MetadataEvent>( projectionOf<MetadataEvent>(Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)))
Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), projection.awaitLoaded()
scope,
)
projection.awaitReady()
val seedList = projection.items val seedList = projection.items
assertEquals(1, seedList.size) assertEquals(1, seedList.size)
val slot = seedList[0] val slot = seedList[0]
@@ -178,7 +180,6 @@ class EventStoreProjectionTest {
awaitFlow(slot) { it.id == v2.id } awaitFlow(slot) { it.id == v2.id }
assertSame(seedList, projection.items, "replaceable update must not change list reference") assertSame(seedList, projection.items, "replaceable update must not change list reference")
assertSame(slot, projection.items[0]) assertSame(slot, projection.items[0])
projection.close()
} }
@Test @Test
@@ -189,15 +190,14 @@ class EventStoreProjectionTest {
observable.insert(v1) observable.insert(v1)
val projection = val projection =
observable.observe<LongTextNoteEvent>( projectionOf<LongTextNoteEvent>(
Filter( Filter(
kinds = listOf(LongTextNoteEvent.KIND), kinds = listOf(LongTextNoteEvent.KIND),
authors = listOf(v1.pubKey), authors = listOf(v1.pubKey),
tags = mapOf("d" to listOf("blog")), tags = mapOf("d" to listOf("blog")),
), ),
scope,
) )
projection.awaitReady() projection.awaitLoaded()
val seedList = projection.items val seedList = projection.items
val slot = seedList[0] val slot = seedList[0]
@@ -206,7 +206,6 @@ class EventStoreProjectionTest {
awaitFlow(slot) { it.id == v2.id } awaitFlow(slot) { it.id == v2.id }
assertSame(seedList, projection.items, "addressable update must not change list reference") assertSame(seedList, projection.items, "addressable update must not change list reference")
projection.close()
} }
/** /**
@@ -228,11 +227,8 @@ class EventStoreProjectionTest {
observable.insert(v2) observable.insert(v2)
val projection = val projection =
observable.observe<MetadataEvent>( projectionOf<MetadataEvent>(Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)))
Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), projection.awaitLoaded()
scope,
)
projection.awaitReady()
val slot = projection.items[0] val slot = projection.items[0]
assertEquals(v2.id, slot.value.id) assertEquals(v2.id, slot.value.id)
@@ -247,7 +243,6 @@ class EventStoreProjectionTest {
delay(150) delay(150)
assertEquals(v2.id, slot.value.id) assertEquals(v2.id, slot.value.id)
projection.close()
} }
@Test @Test
@@ -258,8 +253,8 @@ class EventStoreProjectionTest {
observable.insert(a) observable.insert(a)
observable.insert(b) observable.insert(b)
val projection = observable.observe<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
assertEquals(2, projection.items.size) assertEquals(2, projection.items.size)
val deletion = signer.sign(DeletionEvent.build(listOf(a))) val deletion = signer.sign(DeletionEvent.build(listOf(a)))
@@ -267,7 +262,6 @@ class EventStoreProjectionTest {
val after = projection.awaitItems { it.size == 1 } val after = projection.awaitItems { it.size == 1 }
assertEquals(b.id, after[0].value.id) assertEquals(b.id, after[0].value.id)
projection.close()
} }
/** /**
@@ -280,8 +274,8 @@ class EventStoreProjectionTest {
val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) val a = signer.sign(TextNoteEvent.build("a", createdAt = 100))
observable.insert(a) observable.insert(a)
val projection = observable.observe<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
val seed = projection.items val seed = projection.items
assertEquals(1, seed.size) assertEquals(1, seed.size)
@@ -296,7 +290,6 @@ class EventStoreProjectionTest {
projection.items[0] projection.items[0]
.value.id, .value.id,
) )
projection.close()
} }
@Test @Test
@@ -308,8 +301,8 @@ class EventStoreProjectionTest {
observable.insert(a) observable.insert(a)
observable.insert(b) observable.insert(b)
val projection = observable.observe<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
assertEquals(2, projection.items.size) assertEquals(2, projection.items.size)
val vanish = val vanish =
@@ -323,7 +316,6 @@ class EventStoreProjectionTest {
val after = projection.awaitItems { it.isEmpty() } val after = projection.awaitItems { it.isEmpty() }
assertTrue(after.isEmpty()) assertTrue(after.isEmpty())
projection.close()
} }
/** /**
@@ -337,8 +329,8 @@ class EventStoreProjectionTest {
val a = signer.sign(TextNoteEvent.build("a", createdAt = time)) val a = signer.sign(TextNoteEvent.build("a", createdAt = time))
observable.insert(a) observable.insert(a)
val projection = observable.observe<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) val projection = projectionOf<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)))
projection.awaitReady() projection.awaitLoaded()
val seed = projection.items val seed = projection.items
val foreignVanish = val foreignVanish =
@@ -352,7 +344,6 @@ class EventStoreProjectionTest {
delay(150) delay(150)
assertSame(seed, projection.items) assertSame(seed, projection.items)
projection.close()
} }
/** /**
@@ -370,11 +361,8 @@ class EventStoreProjectionTest {
observable.insert(short) observable.insert(short)
val projection = val projection =
observable.observe<Event>( projectionOf<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)))
Filter(kinds = listOf(TextNoteEvent.KIND)), projection.awaitLoaded()
scope,
)
projection.awaitReady()
assertEquals(2, projection.items.size) assertEquals(2, projection.items.size)
// Let the short expiration lapse, then ask the store to // Let the short expiration lapse, then ask the store to
@@ -385,7 +373,6 @@ class EventStoreProjectionTest {
val after = projection.awaitItems { it.size == 1 } val after = projection.awaitItems { it.size == 1 }
assertEquals(safe.id, after[0].value.id) assertEquals(safe.id, after[0].value.id)
projection.close()
} }
/** /**
@@ -404,11 +391,8 @@ class EventStoreProjectionTest {
observable.insert(foreign) observable.insert(foreign)
val projection = val projection =
observable.observe<Event>( projectionOf<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)))
Filter(kinds = listOf(TextNoteEvent.KIND)), projection.awaitLoaded()
scope,
)
projection.awaitReady()
assertEquals(3, projection.items.size) assertEquals(3, projection.items.size)
// Drop everything authored by `signer` — should leave // Drop everything authored by `signer` — should leave
@@ -417,7 +401,6 @@ class EventStoreProjectionTest {
val after = projection.awaitItems { it.size == 1 } val after = projection.awaitItems { it.size == 1 }
assertEquals(foreign.id, after[0].value.id) assertEquals(foreign.id, after[0].value.id)
projection.close()
} }
@Test @Test
@@ -429,11 +412,8 @@ class EventStoreProjectionTest {
observable.insert(b) observable.insert(b)
val projection = val projection =
observable.observe<TextNoteEvent>( projectionOf<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2))
Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2), projection.awaitLoaded()
scope,
)
projection.awaitReady()
assertEquals(2, projection.items.size) assertEquals(2, projection.items.size)
val c = signer.sign(TextNoteEvent.build("c", createdAt = 300)) val c = signer.sign(TextNoteEvent.build("c", createdAt = 300))
@@ -443,7 +423,6 @@ class EventStoreProjectionTest {
assertEquals(2, after.size) assertEquals(2, after.size)
assertEquals(c.id, after[0].value.id) assertEquals(c.id, after[0].value.id)
assertEquals(b.id, after[1].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 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 filterB = Filter(kinds = listOf(TextNoteEvent.KIND), authors = listOf(authorB.pubKey), limit = 2)
val projection = val projection =
observable.observe<TextNoteEvent>( projectionOf<TextNoteEvent>(
listOf(filterA, filterB), listOf(filterA, filterB),
scope,
) )
projection.awaitReady() projection.awaitLoaded()
assertEquals(4, projection.items.size, "per-filter caps don't dedupe union") assertEquals(4, projection.items.size, "per-filter caps don't dedupe union")
projection.close()
} }
/** /**
@@ -491,11 +468,8 @@ class EventStoreProjectionTest {
observable.insert(a2) observable.insert(a2)
val projection = val projection =
observable.observe<TextNoteEvent>( projectionOf<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2))
Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2), projection.awaitLoaded()
scope,
)
projection.awaitReady()
assertEquals(2, projection.items.size) assertEquals(2, projection.items.size)
observable.insert(a3) observable.insert(a3)
@@ -503,22 +477,33 @@ class EventStoreProjectionTest {
assertEquals(2, after.size) assertEquals(2, after.size)
assertEquals(a3.id, after[0].value.id) assertEquals(a3.id, after[0].value.id)
assertEquals(a2.id, after[1].value.id) assertEquals(a2.id, after[1].value.id)
projection.close()
} }
@Test @Test
fun closeStopsListening() = fun cancellingScopeStopsListening() =
runBlocking { runBlocking {
val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) val a = signer.sign(TextNoteEvent.build("a", createdAt = 100))
observable.insert(a) observable.insert(a)
val projection = observable.observe<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) // Sub-scope so we can cancel just the projection's collector
projection.awaitReady() // without taking down the test's outer scope.
projection.close() val collectorScope = CoroutineScope(SupervisorJob())
val projection =
observable
.project<TextNoteEvent>(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))) observable.insert(signer.sign(TextNoteEvent.build("b", createdAt = 200)))
delay(150) 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 { runBlocking {
val ephemeralKind = 22_000 val ephemeralKind = 22_000
val projection = val projection =
observable.observe<Event>(Filter(kinds = listOf(ephemeralKind)), scope) projectionOf<Event>(Filter(kinds = listOf(ephemeralKind)))
projection.awaitReady() projection.awaitLoaded()
assertTrue(projection.items.isEmpty()) assertTrue(projection.items.isEmpty())
val ephemeral: Event = val ephemeral: Event =
@@ -557,11 +542,8 @@ class EventStoreProjectionTest {
// A fresh projection on the same store gets nothing — the // A fresh projection on the same store gets nothing — the
// event was only ever live, not durable. // event was only ever live, not durable.
val freshProjection = val freshProjection =
observable.observe<Event>(Filter(kinds = listOf(ephemeralKind)), scope) projectionOf<Event>(Filter(kinds = listOf(ephemeralKind)))
freshProjection.awaitReady() freshProjection.awaitLoaded()
assertTrue(freshProjection.items.isEmpty()) assertTrue(freshProjection.items.isEmpty())
projection.close()
freshProjection.close()
} }
} }