refactor(quartz): rename StoreEvent → StoreChange, events flow → changes
The "Event" name was overloaded — the store-mutation type lived
right next to Nostr Event, so StoreEvent.Insert(event: Event) read
awkwardly. Renaming to StoreChange disambiguates and matches
reactive vocabulary ("store changes").
- StoreEvent → StoreChange (file, sealed type, all references).
- ObservableEventStore.events → ObservableEventStore.changes.
- Internal field _events → _changes.
- KDoc cross-references updated.
Case names (Insert, DeleteByFilter, DeleteExpired) unchanged. All
240 store + projection + interner tests still pass.
https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
This commit is contained in:
+10
-10
@@ -63,10 +63,10 @@ import kotlinx.coroutines.yield
|
||||
*
|
||||
* The seed is materialised by querying the store once at start, after
|
||||
* which the projection is driven entirely by
|
||||
* [ObservableEventStore.events]. Three kinds of mutation arrive on
|
||||
* [ObservableEventStore.changes]. Three kinds of mutation arrive on
|
||||
* that stream:
|
||||
*
|
||||
* - [StoreEvent.Insert] — interpreted in-projection so a single
|
||||
* - [StoreChange.Insert] — interpreted in-projection so 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
|
||||
@@ -82,11 +82,11 @@ import kotlinx.coroutines.yield
|
||||
* already lapsed at the moment they arrive are dropped before
|
||||
* they ever enter [items].
|
||||
*
|
||||
* - [StoreEvent.DeleteByFilter] — emitted on `delete(filter)` /
|
||||
* - [StoreChange.DeleteByFilter] — emitted on `delete(filter)` /
|
||||
* `delete(filters)`. The projection drops every slot matching any
|
||||
* of the rule's filters via [Filter.match].
|
||||
*
|
||||
* - [StoreEvent.DeleteExpired] — emitted on `deleteExpiredEvents()`.
|
||||
* - [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
|
||||
@@ -101,7 +101,7 @@ import kotlinx.coroutines.yield
|
||||
* filter under cap, it stays under cap until another match arrives.
|
||||
*
|
||||
* Ephemeral events (kinds `20000-29999`) reach the projection via
|
||||
* [ObservableEventStore.events] without ever being persisted; they
|
||||
* [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
|
||||
@@ -147,7 +147,7 @@ class EventStoreProjection<T : Event>(
|
||||
// emissions arriving during seed and drains them once collect
|
||||
// proceeds. Doing the seed inside `collect { }` would race
|
||||
// with concurrent inserts.
|
||||
store.events
|
||||
store.changes
|
||||
.onSubscription {
|
||||
seed()
|
||||
ready.complete(Unit)
|
||||
@@ -167,20 +167,20 @@ class EventStoreProjection<T : Event>(
|
||||
publish()
|
||||
}
|
||||
|
||||
private fun apply(storeEvent: StoreEvent) {
|
||||
private fun apply(storeEvent: StoreChange) {
|
||||
val changed =
|
||||
when (storeEvent) {
|
||||
is StoreEvent.Insert -> {
|
||||
is StoreChange.Insert -> {
|
||||
applyInsert(storeEvent.event)
|
||||
}
|
||||
|
||||
is StoreEvent.DeleteByFilter -> {
|
||||
is StoreChange.DeleteByFilter -> {
|
||||
dropWhere { ev -> storeEvent.filters.any { it.match(ev) } }
|
||||
}
|
||||
|
||||
// Store's sweep uses strict `<`; isExpirationBefore is
|
||||
// `<=`, so subtract 1 to match.
|
||||
is StoreEvent.DeleteExpired -> {
|
||||
is StoreChange.DeleteExpired -> {
|
||||
val cutoff = (storeEvent.asOf ?: nowProvider()) - 1
|
||||
dropWhere { it.isExpirationBefore(cutoff) }
|
||||
}
|
||||
|
||||
+14
-14
@@ -44,26 +44,26 @@ import kotlinx.coroutines.flow.asSharedFlow
|
||||
* - **Non-ephemeral events** are forwarded to the inner store. If the
|
||||
* inner store rejects (expired, NIP-09 / NIP-62 tombstone, NIP-01
|
||||
* supersession loser), the rejection propagates and nothing is
|
||||
* emitted on [events].
|
||||
* emitted on [changes].
|
||||
* - **Ephemeral events** (kinds `20000-29999`) skip the inner store
|
||||
* entirely — they're never persisted — but they still emit on
|
||||
* [events] so projections can render them while they live. Already
|
||||
* [changes] so projections can render them while they live. Already
|
||||
* expired ephemerals are silently dropped.
|
||||
*
|
||||
* Wrap any store you want to observe — [SQLiteEventStore], FS-backed,
|
||||
* an in-memory test fake — and feed [EventStoreProjection] from the
|
||||
* resulting [events] flow.
|
||||
* resulting [changes] flow.
|
||||
*
|
||||
* Reads (`query`, `count`) and out-of-band writes (`delete`,
|
||||
* `deleteExpiredEvents`) forward to the inner store unchanged. The
|
||||
* latter are *not* surfaced on [events] — see the projection's
|
||||
* latter are *not* surfaced on [changes] — see the projection's
|
||||
* docstring for the rationale.
|
||||
*/
|
||||
class ObservableEventStore(
|
||||
val inner: IEventStore,
|
||||
) : IEventStore {
|
||||
private val _events =
|
||||
MutableSharedFlow<StoreEvent>(
|
||||
private val _changes =
|
||||
MutableSharedFlow<StoreChange>(
|
||||
replay = 0,
|
||||
extraBufferCapacity = 256,
|
||||
onBufferOverflow = BufferOverflow.SUSPEND,
|
||||
@@ -77,9 +77,9 @@ class ObservableEventStore(
|
||||
* transactions emit nothing.
|
||||
*
|
||||
* Projections consume this stream — see [EventStoreProjection]
|
||||
* for how each [StoreEvent] is interpreted.
|
||||
* for how each [StoreChange] is interpreted.
|
||||
*/
|
||||
val events: SharedFlow<StoreEvent> = _events.asSharedFlow()
|
||||
val changes: SharedFlow<StoreChange> = _changes.asSharedFlow()
|
||||
|
||||
override suspend fun insert(event: Event) {
|
||||
if (event.kind.isEphemeral()) {
|
||||
@@ -87,13 +87,13 @@ class ObservableEventStore(
|
||||
// that are already expired — they were never going to live
|
||||
// long enough for a UI to render them.
|
||||
if (event.isExpired()) return
|
||||
_events.emit(StoreEvent.Insert(event))
|
||||
_changes.emit(StoreChange.Insert(event))
|
||||
return
|
||||
}
|
||||
// Non-ephemeral: let the inner store enforce expiration,
|
||||
// tombstones, supersession, etc. If it throws, we never emit.
|
||||
inner.insert(event)
|
||||
_events.emit(StoreEvent.Insert(event))
|
||||
_changes.emit(StoreChange.Insert(event))
|
||||
}
|
||||
|
||||
override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) {
|
||||
@@ -120,7 +120,7 @@ class ObservableEventStore(
|
||||
}
|
||||
// Emit only after the inner transaction commits. If it throws
|
||||
// / rolls back, `accepted` is discarded.
|
||||
for (e in accepted) _events.emit(StoreEvent.Insert(e))
|
||||
for (e in accepted) _changes.emit(StoreChange.Insert(e))
|
||||
}
|
||||
|
||||
override suspend fun <T : Event> query(filter: Filter): List<T> = inner.query(filter)
|
||||
@@ -143,12 +143,12 @@ class ObservableEventStore(
|
||||
|
||||
override suspend fun delete(filter: Filter) {
|
||||
inner.delete(filter)
|
||||
_events.emit(StoreEvent.DeleteByFilter(listOf(filter)))
|
||||
_changes.emit(StoreChange.DeleteByFilter(listOf(filter)))
|
||||
}
|
||||
|
||||
override suspend fun delete(filters: List<Filter>) {
|
||||
inner.delete(filters)
|
||||
_events.emit(StoreEvent.DeleteByFilter(filters))
|
||||
_changes.emit(StoreChange.DeleteByFilter(filters))
|
||||
}
|
||||
|
||||
override suspend fun deleteExpiredEvents() {
|
||||
@@ -159,7 +159,7 @@ class ObservableEventStore(
|
||||
// own clock when the event is processed.
|
||||
val asOf = TimeUtils.now()
|
||||
inner.deleteExpiredEvents()
|
||||
_events.emit(StoreEvent.DeleteExpired(asOf))
|
||||
_changes.emit(StoreChange.DeleteExpired(asOf))
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+5
-5
@@ -24,7 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
|
||||
/**
|
||||
* Mutations published by [ObservableEventStore.events]. Projections
|
||||
* Mutations published by [ObservableEventStore.changes]. Projections
|
||||
* react to these to keep their in-memory view in sync with the
|
||||
* underlying store.
|
||||
*
|
||||
@@ -42,16 +42,16 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
* pin the timestamp it actually used, so the projection drops
|
||||
* exactly the events the store dropped.
|
||||
*/
|
||||
sealed interface StoreEvent {
|
||||
sealed interface StoreChange {
|
||||
data class Insert(
|
||||
val event: Event,
|
||||
) : StoreEvent
|
||||
) : StoreChange
|
||||
|
||||
data class DeleteByFilter(
|
||||
val filters: List<Filter>,
|
||||
) : StoreEvent
|
||||
) : StoreChange
|
||||
|
||||
data class DeleteExpired(
|
||||
val asOf: Long? = null,
|
||||
) : StoreEvent
|
||||
) : StoreChange
|
||||
}
|
||||
+2
-2
@@ -373,7 +373,7 @@ class EventStoreProjectionTest {
|
||||
|
||||
// Let the short expiration lapse, then ask the store to
|
||||
// sweep — the projection drops the expired slot in
|
||||
// response to the resulting StoreEvent.Delete(Expired).
|
||||
// response to the resulting StoreChange.Delete(Expired).
|
||||
delay(2000)
|
||||
observable.deleteExpiredEvents()
|
||||
|
||||
@@ -523,7 +523,7 @@ class EventStoreProjectionTest {
|
||||
* Ephemeral events (kinds `20000-29999`) are never persisted, so
|
||||
* the inner SQLite store silently drops them. The
|
||||
* `ObservableEventStore` wrapper still routes them onto its
|
||||
* [events][com.vitorpamplona.quartz.nip01Core.store.projection.ObservableEventStore.events]
|
||||
* [events][com.vitorpamplona.quartz.nip01Core.store.projection.ObservableEventStore.changes]
|
||||
* flow, so an open projection sees them while it's alive. They
|
||||
* vanish from any future seed because the DB never had them.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user