docs(quartz): trim verbose KDoc on cache + observable layers
Reviews before merge: - EventInterner: drop stale "used by Event.fromJson" reference (we reverted that earlier). Tighten the class doc. - ObservableEventStore: collapse the 30+ line class KDoc + duplicated StoreChange description into one focused doc each. - EventStoreProjection: trim the 60+ line class doc; reactive semantics live on the project() extension. Add explicit caveat about in-place addressable updates not re-evaluating filter membership. No behaviour change. All cache + store + projection tests pass. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
This commit is contained in:
Claude
Vendored
+9
-19
@@ -24,31 +24,21 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
|
||||
/**
|
||||
* Process-wide interner that canonicalises [Event] instances by id so
|
||||
* every consumer (relay client, store deserialization, projections,
|
||||
* tests) sees the same object reference for the same event id.
|
||||
* Canonicalises [Event] instances by id so every consumer sees the
|
||||
* same object reference for the same event id.
|
||||
*
|
||||
* Backed by weak references — entries vanish when no live consumer
|
||||
* holds the event, so the cache only grows as long as projections /
|
||||
* UI state actually need the events. Sized for ~5000 hot events; the
|
||||
* underlying map resizes if usage exceeds that.
|
||||
*
|
||||
* Use [intern] on every event arrival path. The first occurrence wins
|
||||
* and becomes canonical; subsequent equivalent decodes return that
|
||||
* canonical instance.
|
||||
* holds the event. First-seen wins; equivalence is by event id, so
|
||||
* callers must trust ids are content-derived (signed events satisfy
|
||||
* this). Sized for ~5000 hot entries; resizes if usage exceeds that.
|
||||
*
|
||||
* Platforms without weak references fall back to a passthrough that
|
||||
* returns [event] unchanged — no canonicalisation, but no leaks
|
||||
* either.
|
||||
* returns [event] unchanged — no canonicalisation, no leaks.
|
||||
*/
|
||||
expect class EventInterner() {
|
||||
/**
|
||||
* Returns the canonical [Event] for [event]'s id. If a live
|
||||
* canonical instance already exists for this id, returns it;
|
||||
* otherwise stores [event] as the new canonical and returns it.
|
||||
*
|
||||
* Equivalence is by event id only — callers must trust the id
|
||||
* was content-derived (signed events satisfy this).
|
||||
* Returns the canonical [Event] for [event]'s id, storing
|
||||
* [event] as canonical if no live entry exists.
|
||||
*/
|
||||
fun intern(event: Event): Event
|
||||
|
||||
@@ -62,7 +52,7 @@ expect class EventInterner() {
|
||||
fun clear()
|
||||
|
||||
companion object {
|
||||
/** Process-wide default interner used by [Event.fromJson]. */
|
||||
/** Process-wide shared instance. */
|
||||
val Default: EventInterner
|
||||
}
|
||||
}
|
||||
|
||||
+26
-62
@@ -58,70 +58,35 @@ sealed interface ProjectionState<out T : Event> {
|
||||
|
||||
/**
|
||||
* 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:
|
||||
* [ObservableEventStore] for a fixed set of [filters]. Pure logic, no
|
||||
* coroutine ownership: construct, [seed] once, then [apply] each
|
||||
* [StoreChange] and read [snapshot]. For the standard "wrap into a
|
||||
* cold Flow and collect from a ViewModel" path, use
|
||||
* [ObservableEventStore.project] which composes this class.
|
||||
*
|
||||
* - **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 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.
|
||||
* Each visible event is wrapped in a [MutableStateFlow], giving the
|
||||
* UI three change granularities:
|
||||
*
|
||||
* Three kinds of [StoreChange] are interpreted in-projection:
|
||||
* - **Membership** (insert / removal): [snapshot] returns a fresh
|
||||
* list. Stable list reference while membership is unchanged.
|
||||
* - **In-place replaceable / addressable update**: same slot, new
|
||||
* [MutableStateFlow.value]. Only collectors of that handle
|
||||
* re-render. The slot's sort key is frozen at insertion, so the
|
||||
* list does not reshuffle when the new version has a later
|
||||
* `created_at`. *Caveat: the slot stays in whatever filters
|
||||
* matched the original version; if the new version no longer
|
||||
* matches a filter (e.g. its tags changed), the slot is not
|
||||
* re-evaluated and remains live.*
|
||||
* - **Removal**: NIP-09, NIP-62, NIP-40 expiration, `delete(filter)`.
|
||||
*
|
||||
* - [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
|
||||
* `created_at` ties) is honoured.
|
||||
* - **NIP-09 deletions.** A [DeletionEvent] removes any matching
|
||||
* handle owned by the same author (for GiftWrap, the recipient).
|
||||
* Cross-author deletions are inert.
|
||||
* - **NIP-62 right to vanish.** A [RequestToVanishEvent] whose
|
||||
* `shouldVanishFrom(store.relay)` is true drops every handle
|
||||
* 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 the snapshot.
|
||||
* - [StoreChange.DeleteByFilter] — emitted on `delete(filter)` /
|
||||
* `delete(filters)`. Drops every slot matching any of the rule's
|
||||
* filters via [Filter.match].
|
||||
* - [StoreChange.DeleteExpired] — emitted on `deleteExpiredEvents()`.
|
||||
* 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 is **per-filter**: each filter retains at most its own
|
||||
* `limit` matches; the snapshot is the deduped union, so disjoint
|
||||
* matches across filters can exceed any single cap. Removed slots
|
||||
* are not refilled from 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 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 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**: 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.
|
||||
* Ephemeral events appear in the snapshot for as long as the
|
||||
* projection is alive; they never survive a re-seed (the inner store
|
||||
* never had them) and aren't touched by `deleteExpiredEvents()`.
|
||||
*/
|
||||
class EventStoreProjection<T : Event>(
|
||||
private val store: ObservableEventStore,
|
||||
@@ -174,9 +139,8 @@ class EventStoreProjection<T : Event>(
|
||||
dropWhere { ev -> storeEvent.filters.any { it.match(ev) } }
|
||||
}
|
||||
|
||||
// Store's sweep uses strict `<`; isExpirationBefore is
|
||||
// `<=`, so subtract 1 to match.
|
||||
is StoreChange.DeleteExpired -> {
|
||||
// Store's sweep uses strict `<`; isExpirationBefore is `<=`, so subtract 1.
|
||||
val cutoff = (storeEvent.asOf ?: nowProvider()) - 1
|
||||
dropWhere { it.isExpirationBefore(cutoff) }
|
||||
}
|
||||
|
||||
+17
-50
@@ -32,31 +32,18 @@ import kotlinx.coroutines.flow.SharedFlow
|
||||
import kotlinx.coroutines.flow.asSharedFlow
|
||||
|
||||
/**
|
||||
* A reactive façade over any [IEventStore] that publishes every event
|
||||
* accepted for *observation* — a superset of the events the inner
|
||||
* store persists.
|
||||
* Reactive façade over any [IEventStore]. Publishes a [StoreChange]
|
||||
* on [changes] for every accepted mutation so projections can stay in
|
||||
* sync without re-querying.
|
||||
*
|
||||
* The split between persistence and observation is the whole point of
|
||||
* this class:
|
||||
*
|
||||
* - **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 [changes].
|
||||
* - **Ephemeral events** (kinds `20000-29999`) skip the inner store
|
||||
* entirely — they're never persisted — but they still emit on
|
||||
* [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 [changes] flow.
|
||||
*
|
||||
* Reads (`query`, `count`) and out-of-band writes (`delete`,
|
||||
* `deleteExpiredEvents`) forward to the inner store; the latter are
|
||||
* also surfaced on [changes] as [StoreChange.DeleteByFilter] /
|
||||
* [StoreChange.DeleteExpired] so projections can drop the matching
|
||||
* slots in memory without re-querying.
|
||||
* - Non-ephemeral [insert] / [transaction] entries forward to the
|
||||
* inner store. On rejection (expired, NIP-09 / NIP-62 tombstone,
|
||||
* NIP-01 supersession loser) the throw propagates and nothing is
|
||||
* emitted.
|
||||
* - Ephemeral events (kinds `20000-29999`) skip the inner store but
|
||||
* still emit. Already-expired ephemerals are silently dropped.
|
||||
* - [delete] and [deleteExpiredEvents] also emit so projections drop
|
||||
* the matching slots in memory.
|
||||
*/
|
||||
class ObservableEventStore(
|
||||
val inner: IEventStore,
|
||||
@@ -70,35 +57,15 @@ class ObservableEventStore(
|
||||
onBufferOverflow = BufferOverflow.SUSPEND,
|
||||
)
|
||||
|
||||
/**
|
||||
* Stream of mutations accepted by the observable layer. One
|
||||
* emission per successful [insert] (or per accepted event in a
|
||||
* [transaction] body), one emission per [delete] /
|
||||
* [deleteExpiredEvents] call. Rejected inserts and rolled-back
|
||||
* transactions emit nothing.
|
||||
*
|
||||
* Projections consume this stream — see `EventStoreProjection`
|
||||
* for how each [StoreChange] is interpreted.
|
||||
*/
|
||||
/** Stream of mutations accepted by this layer. See [StoreChange] for the cases. */
|
||||
val changes: SharedFlow<StoreChange> = _changes.asSharedFlow()
|
||||
|
||||
/**
|
||||
* Mutations published by [changes]. Projections react to these to
|
||||
* keep their in-memory view in sync with the underlying store.
|
||||
*
|
||||
* - [Insert] is emitted for every event accepted by the
|
||||
* observable layer (persistable or ephemeral). Carries the
|
||||
* event itself so the projection can run its NIP-01 / NIP-09 /
|
||||
* NIP-62 interpretation.
|
||||
* - [DeleteByFilter] is emitted for every `delete(filter)` /
|
||||
* `delete(filters)` call on the observable. Carries the same
|
||||
* filters the store used so projections can apply
|
||||
* [Filter.match] in memory and drop the matching slots without
|
||||
* re-querying.
|
||||
* - [DeleteExpired] is emitted for every `deleteExpiredEvents()`
|
||||
* sweep. The optional [DeleteExpired.asOf] cutoff lets the
|
||||
* store pin the timestamp it actually used, so the projection
|
||||
* drops exactly the events the store dropped.
|
||||
* One [Insert] per accepted event; one [DeleteByFilter] per
|
||||
* `delete(filter[s])`; one [DeleteExpired] per
|
||||
* `deleteExpiredEvents()` sweep. The cutoff carried by
|
||||
* [DeleteExpired] is pinned at the moment the sweep ran so
|
||||
* projections drop exactly the events the store dropped.
|
||||
*/
|
||||
sealed interface StoreChange {
|
||||
data class Insert(
|
||||
|
||||
Reference in New Issue
Block a user